node:stream

The node:stream module provides the Node.js abstract interface for streaming data: the Readable, Writable, Duplex, Transform, and PassThrough classes. A stream reads or writes data as a continuous flow of chunks, so a function processes a large volume of data without holding all of it in memory at once.

Under azion dev, import { pipeline } from "node:stream" fails the build with No matching export in "internal-env-dev:stream" for import "pipeline". The two examples that import pipeline run only in a deployed function.

Examples

Each example is a complete function that imports from node:stream. The response below each example is the one a deployed function returns.

Basic readable and writable streams

This function creates a readable stream and a writable stream, connects them with pipe(), and responds when the pipe is set up:

JavaScript

01/**

02 * An example of using Node.js Stream API in an Azion Function.

03 * Support:

04 * - Partially supported (Extended by library `stream-browserify`)

05 * @module runtime-apis/nodejs/stream/main

06 * @example

07 * // Build and run with the Azion CLI:

08 * azion build

09 * azion dev

10 */

11import stream from "node:stream";

13/**

14 * An example of using the Node.js Stream API in an Azion Function.

15 * @param {*} event

16 * @returns {Promise<Response>}

17 */

18const main = async (event) => {

19 return new Promise((resolve, reject) => {

20 const chunks = ["chunk1", "chunk2", "chunk3", "chunk4", "chunk5"];

22 const nextChunk = () => {

23 const chunk = chunks.shift();

24 if (chunk) {

25 console.log("Chunk", chunk);

26 nextChunk();

27 }

28 };

30 const readable = new stream.Readable({

31 encoding: "utf8",

32 read() {

33 nextChunk();

34 },

35 });

37 const writable = new stream.Writable({

38 write(chunk, encoding, callback) {

39 console.log("Chunk", chunk.toString());

40 callback();

41 },

42 });

44 readable.pipe(writable);

46 resolve(new Response("Done"));

47 });

48};

50export default main;

The function responds with:

Done

Transform stream for data processing

A Transform stream changes data as it passes through. This function connects two transforms and a collector with pipeline(): the first uppercases each line, the second numbers it:

JavaScript

01import { Transform, pipeline } from "node:stream";

03const main = async (event) => {

04 // Create a transform stream that uppercases text

05 const upperCaseTransform = new Transform({

06 transform(chunk, encoding, callback) {

07 const uppercased = chunk.toString().toUpperCase();

08 this.push(uppercased);

09 callback();

10 }

11 });

13 // Create a transform stream that adds line numbers

14 let lineNumber = 0;

15 const addLineNumbers = new Transform({

16 transform(chunk, encoding, callback) {

17 lineNumber++;

18 const numbered = `${lineNumber}: ${chunk.toString()}`;

19 this.push(numbered);

20 callback();

21 }

22 });

24 // Collect output

25 const output = [];

26 const collector = new Transform({

27 transform(chunk, encoding, callback) {

28 output.push(chunk.toString());

29 callback();

30 }

31 });

33 // Simulate input data

34 const inputData = ["hello\n", "world\n", "azion\n", "runtime\n"];

36 // Process each chunk through the pipeline

37 for (const data of inputData) {

38 upperCaseTransform.write(data);

39 }

40 upperCaseTransform.end();

42 return new Promise((resolve) => {

43 pipeline(

44 upperCaseTransform,

45 addLineNumbers,

46 collector,

47 (err) => {

48 if (err) {

49 console.error("Pipeline error:", err);

50 }

51 console.log("Output:", output.join(""));

52 resolve(new Response(output.join("")));

53 }

54 );

55 });

56};

58export default main;

The function responds with the numbered, uppercased lines:

Text

011: HELLO

022: WORLD

033: AZION

044: RUNTIME

PassThrough stream for data forwarding

A PassThrough stream forwards data without changing it. This function writes three chunks to one and collects each chunk as it passes:

JavaScript

01import { PassThrough, pipeline } from "node:stream";

03const main = async (event) => {

04 // Create a PassThrough stream

05 const passThrough = new PassThrough();

07 // Collect data that passes through

08 const collected = [];

09 passThrough.on("data", (chunk) => {

10 collected.push(chunk.toString());

11 console.log("Data passed through:", chunk.toString());

12 });

14 // Write data to the stream

15 const data = ["First chunk\n", "Second chunk\n", "Third chunk\n"];

16 data.forEach(chunk => passThrough.write(chunk));

17 passThrough.end();

19 // Wait for all data to be processed

20 await new Promise(resolve => passThrough.on("finish", resolve));

22 return new Response(JSON.stringify({

23 message: "Data passed through successfully",

24 chunks: collected,

25 totalChunks: collected.length

26 }), {

27 headers: { "Content-Type": "application/json" }

28 });

29};

31export default main;

The function responds with the chunks it collected:

{"message":"Data passed through successfully","chunks":["First chunk\n","Second chunk\n","Third chunk\n"],"totalChunks":3}

Duplex stream for bidirectional communication

A Duplex stream is both readable and writable. This function writes two chunks to a duplex stream and reads three chunks from it:

JavaScript

01import { Duplex } from "node:stream";

03const main = async (event) => {

04 // Create a duplex stream

05 const duplex = new Duplex({

06 read(size) {

07 // Simulate reading data

08 if (this._readCounter === undefined) this._readCounter = 0;

09 this._readCounter++;

11 if (this._readCounter <= 3) {

12 this.push(`Read chunk ${this._readCounter}\n`);

13 } else {

14 this.push(null); // End the stream

15 }

16 },

17 write(chunk, encoding, callback) {

18 console.log("Written:", chunk.toString().trim());

19 callback();

20 }

21 });

23 // Write to the duplex stream

24 duplex.write("Data to write\n");

25 duplex.write("More data\n");

27 // Read from the duplex stream

28 const output = [];

29 duplex.on("data", (chunk) => {

30 output.push(chunk.toString());

31 });

33 // Wait for stream to end

34 await new Promise(resolve => duplex.on("end", resolve));

36 return new Response(JSON.stringify({

37 readOutput: output.join(""),

38 message: "Duplex stream processed"

39 }), {

40 headers: { "Content-Type": "application/json" }

41 });

42};

44export default main;

The function logs Written: Data to write and Written: More data, and responds with the chunks it read:

{"readOutput":"Read chunk 1\nRead chunk 2\nRead chunk 3\n","message":"Duplex stream processed"}

Request body processing with streams

This function reads the request body chunk by chunk, passes each chunk through two transforms, and counts the bytes:

JavaScript

01import { Transform } from "node:stream";

02import { Buffer } from "node:buffer";

04const main = async (event) => {

05 const request = event.request;

07 // Create a transform stream to process chunks

08 let totalBytes = 0;

09 const byteCounter = new Transform({

10 transform(chunk, encoding, callback) {

11 totalBytes += chunk.length;

12 this.push(chunk);

13 callback();

14 }

15 });

17 // Create a transform to collect data

18 const chunks = [];

19 const collector = new Transform({

20 transform(chunk, encoding, callback) {

21 chunks.push(chunk);

22 this.push(chunk);

23 callback();

24 }

25 });

27 // Get request body as stream (if available)

28 if (request.body) {

29 const reader = request.body.getReader();

31 // Process the stream

32 while (true) {

33 const { done, value } = await reader.read();

34 if (done) break;

36 byteCounter.write(Buffer.from(value));

37 collector.write(Buffer.from(value));

38 }

40 byteCounter.end();

41 collector.end();

42 }

44 // Combine collected chunks

45 const bodyContent = chunks.length > 0

46 ? Buffer.concat(chunks).toString("utf8")

47 : "No body content";

49 return new Response(JSON.stringify({

50 totalBytes,

51 chunkCount: chunks.length,

52 bodyPreview: bodyContent.substring(0, 200)

53 }), {

54 headers: { "Content-Type": "application/json" }

55 });

56};

58export default main;

For a POST request whose body is streamed request body for the stream sample, the function responds with the byte count, the number of chunks, and the start of the body:

{"totalBytes":43,"chunkCount":1,"bodyPreview":"streamed request body for the stream sample"}

Supported APIs

The table lists the status of each node:stream API in Azion Runtime:

API	Status
`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

stream.pipeline() takes a callback, as the transform example shows. The promise form, pipeline() from node:stream/promises, throws Error: [unenv] stream.promises.pipeline is not implemented yet! in a deployed function and under azion dev.

Node.js APIs

The status of every Node.js module in Azion Runtime, stream included.

Use Node.js APIs through polyfills

How the build turns node: imports into code that runs in Azion Runtime.

ReadableStream

The Web Streams alternative for reading data in chunks.

Node.js stream documentation

The full Node.js reference for every node:stream API in the table.