node:stream

O módulo node:stream no Azion Runtime: a interface abstrata para dados em streaming, exemplos de cada tipo de stream e as APIs suportadas.

Última atualização

Ver como Markdown Configurar agente

O módulo node:stream fornece a interface abstrata do Node.js para dados em streaming: as classes Readable, Writable, Duplex, Transform e PassThrough. Um stream lê ou grava dados como um fluxo contínuo de chunks, então uma function processa um grande volume de dados sem manter todos eles na memória ao mesmo tempo.

nota — Com azion dev, import { pipeline } from "node:stream" faz o build falhar com No matching export in "internal-env-dev:stream" for import "pipeline". Os dois exemplos que importam pipeline são executados apenas em uma function após o deploy.

Exemplos

Cada exemplo é uma function completa que importa de node:stream. A resposta abaixo de cada exemplo é a que uma function com deploy feito retorna.

Streams básicos de leitura e escrita

Esta function cria um stream de leitura e um stream de escrita, conecta os dois com pipe() e responde quando o pipe está configurado:

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";
12
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"];
21
22 const nextChunk = () => {
23 const chunk = chunks.shift();
24 if (chunk) {
25 console.log("Chunk", chunk);
26 nextChunk();
27 }
28 };
29
30 const readable = new stream.Readable({
31 encoding: "utf8",
32 read() {
33 nextChunk();
34 },
35 });
36
37 const writable = new stream.Writable({
38 write(chunk, encoding, callback) {
39 console.log("Chunk", chunk.toString());
40 callback();
41 },
42 });
43
44 readable.pipe(writable);
45
46 resolve(new Response("Done"));
47 });
48};
49
50export default main;

A function responde com:

Done

Stream Transform para processamento de dados

Um stream Transform altera os dados à medida que eles passam por ele. Esta function conecta dois transforms e um coletor com pipeline(): o primeiro converte cada linha para maiúsculas e o segundo numera cada linha:

JavaScript
01import { Transform, pipeline } from "node:stream";
02
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 });
12
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 });
23
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 });
32
33 // Simulate input data
34 const inputData = ["hello\n", "world\n", "azion\n", "runtime\n"];
35
36 // Process each chunk through the pipeline
37 for (const data of inputData) {
38 upperCaseTransform.write(data);
39 }
40 upperCaseTransform.end();
41
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};
57
58export default main;

A function responde com as linhas numeradas e em maiúsculas:

Texto
011: HELLO
022: WORLD
033: AZION
044: RUNTIME

Stream PassThrough para encaminhamento de dados

Um stream PassThrough encaminha dados sem alterá-los. Esta function grava três chunks em um stream desse tipo e coleta cada chunk à medida que ele passa:

JavaScript
01import { PassThrough, pipeline } from "node:stream";
02
03const main = async (event) => {
04 // Create a PassThrough stream
05 const passThrough = new PassThrough();
06
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 });
13
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();
18
19 // Wait for all data to be processed
20 await new Promise(resolve => passThrough.on("finish", resolve));
21
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};
30
31export default main;

A function responde com os chunks que coletou:

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

Stream Duplex para comunicação bidirecional

Um stream Duplex é ao mesmo tempo de leitura e de escrita. Esta function grava dois chunks em um stream duplex e lê três chunks dele:

JavaScript
01import { Duplex } from "node:stream";
02
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++;
10
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 });
22
23 // Write to the duplex stream
24 duplex.write("Data to write\n");
25 duplex.write("More data\n");
26
27 // Read from the duplex stream
28 const output = [];
29 duplex.on("data", (chunk) => {
30 output.push(chunk.toString());
31 });
32
33 // Wait for stream to end
34 await new Promise(resolve => duplex.on("end", resolve));
35
36 return new Response(JSON.stringify({
37 readOutput: output.join(""),
38 message: "Duplex stream processed"
39 }), {
40 headers: { "Content-Type": "application/json" }
41 });
42};
43
44export default main;

A function registra Written: Data to write e Written: More data no log e responde com os chunks que leu:

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

Processamento do corpo da requisição com streams

Esta function lê o corpo da requisição chunk por chunk, passa cada chunk por dois transforms e conta os bytes:

JavaScript
01import { Transform } from "node:stream";
02import { Buffer } from "node:buffer";
03
04const main = async (event) => {
05 const request = event.request;
06
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 });
16
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 });
26
27 // Get request body as stream (if available)
28 if (request.body) {
29 const reader = request.body.getReader();
30
31 // Process the stream
32 while (true) {
33 const { done, value } = await reader.read();
34 if (done) break;
35
36 byteCounter.write(Buffer.from(value));
37 collector.write(Buffer.from(value));
38 }
39
40 byteCounter.end();
41 collector.end();
42 }
43
44 // Combine collected chunks
45 const bodyContent = chunks.length > 0
46 ? Buffer.concat(chunks).toString("utf8")
47 : "No body content";
48
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};
57
58export default main;

Para uma requisição POST cujo corpo é streamed request body for the stream sample, a function responde com a contagem de bytes, o número de chunks e o início do corpo:

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

APIs suportadas

A tabela lista o status de cada API de node:stream no Azion Runtime:

APIStatus
stream.Readable🟢 Com suporte
stream.Writable🟢 Com suporte
stream.Duplex🟢 Com suporte
stream.Transform🟢 Com suporte
stream.PassThrough🟢 Com suporte
stream.pipeline()🟢 Com suporte
stream.compose()🟡 Parcialmente suportado
stream.finished()🟡 Parcialmente suportado
readable.pipe()🟢 Com suporte
readable.on('data')🟢 Com suporte
readable.on('end')🟢 Com suporte
readable.on('error')🟢 Com suporte
writable.write()🟢 Com suporte
writable.end()🟢 Com suporte
transform.push()🟢 Com suporte

stream.pipeline() recebe um callback, como mostra o exemplo com Transform. A forma com promise, pipeline() de node:stream/promises, lança Error: [unenv] stream.promises.pipeline is not implemented yet! em uma function após o deploy e com azion dev.

Recursos relacionados

APIs do Node.js

O status de cada módulo do Node.js no Azion Runtime, incluindo stream.

Use APIs do Node.js com polyfills

Como o build transforma imports node: em código que é executado no Azion Runtime.

ReadableStream

A alternativa da Web Streams API para ler dados em chunks.

Documentação de stream do Node.js

A referência completa do Node.js para cada API de node:stream da tabela.