Node.js API compatibility - Stream

This documentation explores the integration and use of the Node.js Stream module within the Azion Runtime. The Stream module is a fundamental part of Node.js, providing an abstract interface for handling streaming data.

View as Markdown

The stream module in Node.js is a core component that facilitates the handling of streaming data. It allows developers to read and write data in a continuous flow, making it efficient for processing large volumes of data without consuming excessive memory.

Example: Basic readable and writable streams

The example below shows how to use the stream module in a function:

/**
* An example of using Node.js Stream API in an Azion Function.
* Support:
* - Partially supported (Extended by library `stream-browserify`)
* @module runtime-apis/nodejs/stream/main
* @example
* // Execute with Azion Bundler:
* npx edge-functions build
* npx edge-functions dev
*/
import stream from "node:stream";
/**
* An example of using the Node.js Stream API in an Azion Function.
* @param {*} event
* @returns {Promise<Response>}
*/
const main = async (event) => {
return new Promise((resolve, reject) => {
const chunks = ["chunk1", "chunk2", "chunk3", "chunk4", "chunk5"];
const nextChunk = () => {
const chunk = chunks.shift();
if (chunk) {
console.log("Chunk", chunk);
nextChunk();
}
};
const readable = new stream.Readable({
encoding: "utf8",
read() {
nextChunk();
},
});
const writable = new stream.Writable({
write(chunk, encoding, callback) {
console.log("Chunk", chunk.toString());
callback();
},
});
readable.pipe(writable);
resolve(new Response("Done"));
});
};
export default main;

Example: Transform stream for data processing

Use Transform streams to modify data as it passes through:

import { Transform, pipeline } from "node:stream";
const main = async (event) => {
// Create a transform stream that uppercases text
const upperCaseTransform = new Transform({
transform(chunk, encoding, callback) {
const uppercased = chunk.toString().toUpperCase();
this.push(uppercased);
callback();
}
});
// Create a transform stream that adds line numbers
let lineNumber = 0;
const addLineNumbers = new Transform({
transform(chunk, encoding, callback) {
lineNumber++;
const numbered = `${lineNumber}: ${chunk.toString()}`;
this.push(numbered);
callback();
}
});
// Collect output
const output = [];
const collector = new Transform({
transform(chunk, encoding, callback) {
output.push(chunk.toString());
callback();
}
});
// Simulate input data
const inputData = ["hello\n", "world\n", "azion\n", "runtime\n"];
// Process each chunk through the pipeline
for (const data of inputData) {
upperCaseTransform.write(data);
}
upperCaseTransform.end();
return new Promise((resolve) => {
pipeline(
upperCaseTransform,
addLineNumbers,
collector,
(err) => {
if (err) {
console.error("Pipeline error:", err);
}
console.log("Output:", output.join(""));
resolve(new Response(output.join("")));
}
);
});
};
export default main;

Example: PassThrough stream for data forwarding

Use PassThrough streams to forward data without modification:

import { PassThrough, pipeline } from "node:stream";
const main = async (event) => {
// Create a PassThrough stream
const passThrough = new PassThrough();
// Collect data that passes through
const collected = [];
passThrough.on("data", (chunk) => {
collected.push(chunk.toString());
console.log("Data passed through:", chunk.toString());
});
// Write data to the stream
const data = ["First chunk\n", "Second chunk\n", "Third chunk\n"];
data.forEach(chunk => passThrough.write(chunk));
passThrough.end();
// Wait for all data to be processed
await new Promise(resolve => passThrough.on("finish", resolve));
return new Response(JSON.stringify({
message: "Data passed through successfully",
chunks: collected,
totalChunks: collected.length
}), {
headers: { "Content-Type": "application/json" }
});
};
export default main;

Example: Duplex stream for bidirectional communication

Create streams that can both read and write:

import { Duplex } from "node:stream";
const main = async (event) => {
// Create a duplex stream
const duplex = new Duplex({
read(size) {
// Simulate reading data
if (this._readCounter === undefined) this._readCounter = 0;
this._readCounter++;
if (this._readCounter <= 3) {
this.push(`Read chunk ${this._readCounter}\n`);
} else {
this.push(null); // End the stream
}
},
write(chunk, encoding, callback) {
console.log("Written:", chunk.toString().trim());
callback();
}
});
// Write to the duplex stream
duplex.write("Data to write\n");
duplex.write("More data\n");
// Read from the duplex stream
const output = [];
duplex.on("data", (chunk) => {
output.push(chunk.toString());
});
// Wait for stream to end
await new Promise(resolve => duplex.on("end", resolve));
return new Response(JSON.stringify({
readOutput: output.join(""),
message: "Duplex stream processed"
}), {
headers: { "Content-Type": "application/json" }
});
};
export default main;

Example: Processing request body with streams

Stream request body data for efficient processing:

import { Transform } from "node:stream";
import { Buffer } from "node:buffer";
const main = async (event) => {
const request = event.request;
// Create a transform stream to process chunks
let totalBytes = 0;
const byteCounter = new Transform({
transform(chunk, encoding, callback) {
totalBytes += chunk.length;
this.push(chunk);
callback();
}
});
// Create a transform to collect data
const chunks = [];
const collector = new Transform({
transform(chunk, encoding, callback) {
chunks.push(chunk);
this.push(chunk);
callback();
}
});
// Get request body as stream (if available)
if (request.body) {
const reader = request.body.getReader();
// Process the stream
while (true) {
const { done, value } = await reader.read();
if (done) break;
byteCounter.write(Buffer.from(value));
collector.write(Buffer.from(value));
}
byteCounter.end();
collector.end();
}
// Combine collected chunks
const bodyContent = chunks.length > 0
? Buffer.concat(chunks).toString("utf8")
: "No body content";
return new Response(JSON.stringify({
totalBytes,
chunkCount: chunks.length,
bodyPreview: bodyContent.substring(0, 200)
}), {
headers: { "Content-Type": "application/json" }
});
};
export default main;

Supported APIs

APIStatus
stream.Readable🟢 Supported
stream.Writable🟢 Supported
stream.Duplex🟢 Supported
stream.Transform🟢 Supported
stream.PassThrough🟢 Supported
stream.pipeline()🟢 Supported
stream.compose()🟡 Partially supported
stream.finished()🟡 Partially supported
readable.pipe()🟢 Supported
readable.on('data')🟢 Supported
readable.on('end')🟢 Supported
readable.on('error')🟢 Supported
writable.write()🟢 Supported
writable.end()🟢 Supported
transform.push()🟢 Supported

Only the APIs Duplex, PassThrough, Readable, Stream, Transform, and Writable are supported.