ClientRequest

import { EventEmitterAsyncResource, EventEmitter } from 'node:events';
import { notStrictEqual, strictEqual } from 'node:assert';
import { executionAsyncId, triggerAsyncId } from 'node:async_hooks';

// Async tracking tooling will identify this as 'Q'.
const ee1 = new EventEmitterAsyncResource({ name: 'Q' });

// 'foo' listeners will run in the EventEmitters async context.
ee1.on('foo', () => {
strictEqual(executionAsyncId(), ee1.asyncId);
strictEqual(triggerAsyncId(), ee1.triggerAsyncId);
});

const ee2 = new EventEmitter();

// 'foo' listeners on ordinary EventEmitters that do not track async
// context, however, run in the same async context as the emit().
ee2.on('foo', () => {
notStrictEqual(executionAsyncId(), ee2.asyncId);
notStrictEqual(triggerAsyncId(), ee2.triggerAsyncId);
});

Promise.resolve().then(() => {
ee1.emit('foo');
ee2.emit('foo');
});

import { EventEmitter } from 'node:events';
const emitter = new EventEmitter();
emitter.setMaxListeners(emitter.getMaxListeners() + 1);
emitter.once('event', () => {
// do stuff
emitter.setMaxListeners(Math.max(emitter.getMaxListeners() - 1, 0));
});

import { pipeline } from 'node:stream';
import fs from 'node:fs';
import zlib from 'node:zlib';

// Use the pipeline API to easily pipe a series of streams
// together and get notified when the pipeline is fully done.

// A pipeline to gzip a potentially huge tar file efficiently:

pipeline(
fs.createReadStream('archive.tar'),
zlib.createGzip(),
fs.createWriteStream('archive.tar.gz'),
(err) => {
if (err) {
console.error('Pipeline failed.', err);
} else {
console.log('Pipeline succeeded.');
}
},
);

import fs from 'node:fs';
import http from 'node:http';
import { pipeline } from 'node:stream';

const server = http.createServer((req, res) => {
const fileStream = fs.createReadStream('./fileNotExist.txt');
pipeline(fileStream, res, (err) => {
if (err) {
console.log(err); // No such file
// this message can't be sent once `pipeline` already destroyed the socket
return res.end('error!!!');
}
});
});

import http from 'node:http';

// Server has a 5 seconds keep-alive timeout by default
http
.createServer((req, res) => {
res.write('hello\n');
res.end();
})
.listen(3000);

setInterval(() => {
// Adapting a keep-alive agent
http.get('http://localhost:3000', { agent }, (res) => {
res.on('data', (data) => {
// Do nothing
});
});
}, 5000); // Sending request on 5s interval so it's easy to hit idle timeout

import http from 'node:http';
const agent = new http.Agent({ keepAlive: true });

function retriableRequest() {
const req = http
.get('http://localhost:3000', { agent }, (res) => {
// ...
})
.on('error', (err) => {
// Check if retry is needed
if (req.reusedSocket && err.code === 'ECONNRESET') {
retriableRequest();
}
});
}

retriableRequest();

// Write 'hello, ' and then end with 'world!'.
import fs from 'node:fs';
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed!

// Write 'hello, ' and then end with 'world!'.
import fs from 'node:fs';
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed!

// Write 'hello, ' and then end with 'world!'.
import fs from 'node:fs';
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed!

import { addAbortListener } from 'node:events';

function example(signal) {
let disposable;
try {
signal.addEventListener('abort', (e) => e.stopImmediatePropagation());
disposable = addAbortListener(signal, (e) => {
// Do something when signal is aborted.
});
} finally {
disposable?.[Symbol.dispose]();
}
}

import fs from 'node:fs';

const controller = new AbortController();
const read = addAbortSignal(
controller.signal,
fs.createReadStream(('object.json')),
);
// Later, abort the operation closing the stream
controller.abort();

const controller = new AbortController();
setTimeout(() => controller.abort(), 10_000); // set a timeout
const stream = addAbortSignal(
controller.signal,
fs.createReadStream(('object.json')),
);
(async () => {
try {
for await (const chunk of stream) {
await process(chunk);
}
} catch (e) {
if (e.name === 'AbortError') {
// The operation was cancelled
} else {
throw e;
}
}
})();

const controller = new AbortController();
const rs = new ReadableStream({
start(controller) {
controller.enqueue('hello');
controller.enqueue('world');
controller.close();
},
});

addAbortSignal(controller.signal, rs);

finished(rs, (err) => {
if (err) {
if (err.name === 'AbortError') {
// The operation was cancelled
}
}
});

const reader = rs.getReader();

reader.read().then(({ value, done }) => {
console.log(value); // hello
console.log(done); // false
controller.abort();
});

addListener(event: "abort", listener: () => void): this

addListener(event: "close", listener: () => void): this

addListener(event: "connect", listener: (response: IncomingMessage, socket: Socket, head: NonSharedBuffer) => void): this

addListener(event: "continue", listener: () => void): this

addListener(event: "drain", listener: () => void): this

addListener(event: "error", listener: (err: Error) => void): this

addListener(event: "finish", listener: () => void): this

addListener(event: "information", listener: (info: InformationEvent) => void): this

addListener(event: "pipe", listener: (src: stream.Readable) => void): this

addListener(event: "response", listener: (response: IncomingMessage) => void): this

addListener(event: "socket", listener: (socket: Socket) => void): this

addListener(event: "timeout", listener: () => void): this

addListener(event: "unpipe", listener: (src: stream.Readable) => void): this

addListener(event: "upgrade", listener: (response: IncomingMessage, socket: Socket, head: NonSharedBuffer) => void): this

const [ sideA, sideB ] = duplexPair();

import { getEventListeners, EventEmitter } from 'node:events';

{
const ee = new EventEmitter();
const listener = () => console.log('Events are fun');
ee.on('foo', listener);
console.log(getEventListeners(ee, 'foo')); // [ [Function: listener] ]
}
{
const et = new EventTarget();
const listener = () => console.log('Events are fun');
et.addEventListener('foo', listener);
console.log(getEventListeners(et, 'foo')); // [ [Function: listener] ]
}

import { getMaxListeners, setMaxListeners, EventEmitter } from 'node:events';

{
const ee = new EventEmitter();
console.log(getMaxListeners(ee)); // 10
setMaxListeners(11, ee);
console.log(getMaxListeners(ee)); // 11
}
{
const et = new EventTarget();
console.log(getMaxListeners(et)); // 10
setMaxListeners(11, et);
console.log(getMaxListeners(et)); // 11
}

request.setHeader('Foo', 'bar');
request.setHeader('Set-Cookie', ['foo=bar', 'bar=baz']);

const headerNames = request.getRawHeaderNames();
// headerNames === ['Foo', 'Set-Cookie']

import { EventEmitter, listenerCount } from 'node:events';

const myEmitter = new EventEmitter();
myEmitter.on('event', () => {});
myEmitter.on('event', () => {});
console.log(listenerCount(myEmitter, 'event'));
// Prints: 2

import { on, EventEmitter } from 'node:events';
import process from 'node:process';

const ee = new EventEmitter();

// Emit later on
process.nextTick(() => {
ee.emit('foo', 'bar');
ee.emit('foo', 42);
});

for await (const event of on(ee, 'foo')) {
// The execution of this inner block is synchronous and it
// processes one event at a time (even with await). Do not use
// if concurrent execution is required.
console.log(event); // prints ['bar'] [42]
}
// Unreachable here

import { on, EventEmitter } from 'node:events';
import process from 'node:process';

const ac = new AbortController();

(async () => {
const ee = new EventEmitter();

// Emit later on
process.nextTick(() => {
ee.emit('foo', 'bar');
ee.emit('foo', 42);
});

for await (const event of on(ee, 'foo', { signal: ac.signal })) {
// The execution of this inner block is synchronous and it
// processes one event at a time (even with await). Do not use
// if concurrent execution is required.
console.log(event); // prints ['bar'] [42]
}
// Unreachable here
})();

process.nextTick(() => ac.abort());

import { on, EventEmitter } from 'node:events';
import process from 'node:process';

const ee = new EventEmitter();

// Emit later on
process.nextTick(() => {
ee.emit('foo', 'bar');
ee.emit('foo', 42);
ee.emit('close');
});

for await (const event of on(ee, 'foo', { close: ['close'] })) {
console.log(event); // prints ['bar'] [42]
}
// the loop will exit after 'close' is emitted
console.log('done'); // prints 'done'

import { on, EventEmitter } from 'node:events';
import process from 'node:process';

const ee = new EventEmitter();

// Emit later on
process.nextTick(() => {
ee.emit('foo', 'bar');
ee.emit('foo', 42);
});

for await (const event of on(ee, 'foo')) {
// The execution of this inner block is synchronous and it
// processes one event at a time (even with await). Do not use
// if concurrent execution is required.
console.log(event); // prints ['bar'] [42]
}
// Unreachable here

import { on, EventEmitter } from 'node:events';
import process from 'node:process';

const ac = new AbortController();

(async () => {
const ee = new EventEmitter();

// Emit later on
process.nextTick(() => {
ee.emit('foo', 'bar');
ee.emit('foo', 42);
});

for await (const event of on(ee, 'foo', { signal: ac.signal })) {
// The execution of this inner block is synchronous and it
// processes one event at a time (even with await). Do not use
// if concurrent execution is required.
console.log(event); // prints ['bar'] [42]
}
// Unreachable here
})();

process.nextTick(() => ac.abort());

import { on, EventEmitter } from 'node:events';
import process from 'node:process';

const ee = new EventEmitter();

// Emit later on
process.nextTick(() => {
ee.emit('foo', 'bar');
ee.emit('foo', 42);
ee.emit('close');
});

for await (const event of on(ee, 'foo', { close: ['close'] })) {
console.log(event); // prints ['bar'] [42]
}
// the loop will exit after 'close' is emitted
console.log('done'); // prints 'done'

on(event: "abort", listener: () => void): this

on(event: "close", listener: () => void): this

on(event: "connect", listener: (response: IncomingMessage, socket: Socket, head: NonSharedBuffer) => void): this

on(event: "continue", listener: () => void): this

on(event: "drain", listener: () => void): this

on(event: "error", listener: (err: Error) => void): this

on(event: "finish", listener: () => void): this

on(event: "information", listener: (info: InformationEvent) => void): this

on(event: "pipe", listener: (src: stream.Readable) => void): this

on(event: "response", listener: (response: IncomingMessage) => void): this

on(event: "timeout", listener: () => void): this

on(event: "unpipe", listener: (src: stream.Readable) => void): this

on(event: "upgrade", listener: (response: IncomingMessage, socket: Socket, head: NonSharedBuffer) => void): this

on(event: "socket", listener: (socket: Socket) => void): this

import { once, EventEmitter } from 'node:events';
import process from 'node:process';

const ee = new EventEmitter();

process.nextTick(() => {
ee.emit('myevent', 42);
});

const [value] = await once(ee, 'myevent');
console.log(value);

const err = new Error('kaboom');
process.nextTick(() => {
ee.emit('error', err);
});

try {
await once(ee, 'myevent');
} catch (err) {
console.error('error happened', err);
}

import { EventEmitter, once } from 'node:events';

const ee = new EventEmitter();

once(ee, 'error')
.then(([err]) => console.log('ok', err.message))
.catch((err) => console.error('error', err.message));

ee.emit('error', new Error('boom'));

// Prints: ok boom

import { EventEmitter, once } from 'node:events';

const ee = new EventEmitter();
const ac = new AbortController();

async function foo(emitter, event, signal) {
try {
await once(emitter, event, { signal });
console.log('event emitted!');
} catch (error) {
if (error.name === 'AbortError') {
console.error('Waiting for the event was canceled!');
} else {
console.error('There was an error', error.message);
}
}
}

foo(ee, 'foo', ac.signal);
ac.abort(); // Abort waiting for the event
ee.emit('foo'); // Prints: Waiting for the event was canceled!

import { once, EventEmitter } from 'node:events';
import process from 'node:process';

const ee = new EventEmitter();

process.nextTick(() => {
ee.emit('myevent', 42);
});

const [value] = await once(ee, 'myevent');
console.log(value);

const err = new Error('kaboom');
process.nextTick(() => {
ee.emit('error', err);
});

try {
await once(ee, 'myevent');
} catch (err) {
console.error('error happened', err);
}

import { EventEmitter, once } from 'node:events';

const ee = new EventEmitter();

once(ee, 'error')
.then(([err]) => console.log('ok', err.message))
.catch((err) => console.error('error', err.message));

ee.emit('error', new Error('boom'));

// Prints: ok boom

import { EventEmitter, once } from 'node:events';

const ee = new EventEmitter();
const ac = new AbortController();

async function foo(emitter, event, signal) {
try {
await once(emitter, event, { signal });
console.log('event emitted!');
} catch (error) {
if (error.name === 'AbortError') {
console.error('Waiting for the event was canceled!');
} else {
console.error('There was an error', error.message);
}
}
}

foo(ee, 'foo', ac.signal);
ac.abort(); // Abort waiting for the event
ee.emit('foo'); // Prints: Waiting for the event was canceled!

once(event: "abort", listener: () => void): this

once(event: "close", listener: () => void): this

once(event: "connect", listener: (response: IncomingMessage, socket: Socket, head: NonSharedBuffer) => void): this

once(event: "continue", listener: () => void): this

once(event: "drain", listener: () => void): this

once(event: "error", listener: (err: Error) => void): this

once(event: "finish", listener: () => void): this

once(event: "information", listener: (info: InformationEvent) => void): this