Compatibilidade APIs Node.js - Stream

Esta documentação explora a integração e uso do módulo Stream do Node.js dentro do Azion Runtime. O módulo Stream é uma parte fundamental do Node.js, fornecendo uma interface abstrata para manipular dados em streaming.

Ver como Markdown

O módulo stream no Node.js é um componente core que facilita a manipulação de dados em streaming. Ele permite que desenvolvedores leiam e escrevam dados em um fluxo contínuo, tornando eficiente o processamento de grandes volumes de dados sem consumir memória excessiva.

Exemplo: Streams legíveis e graváveis básicos

O exemplo abaixo mostra como usar o módulo stream em uma 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;

Exemplo: Transform stream para processamento de dados

Use Transform streams para modificar dados conforme passam pelo stream:

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;

Exemplo: PassThrough stream para encaminhamento de dados

Use PassThrough streams para encaminhar dados sem modificação:

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;

Exemplo: Duplex stream para comunicação bidirecional

Crie streams que podem tanto ler quanto escrever:

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;

Exemplo: Processando corpo da requisição com streams

Processe dados do corpo da requisição de forma eficiente com streams:

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;

APIs com suporte

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

Apenas as APIs Duplex, PassThrough, Readable, Stream, Transform, e Writable têm suporte.

Recursos relacionados