# node:stream

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
/**
 * 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
 * // Build and run with the Azion CLI:
 * azion build
 * azion 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;
```

A function responde com:

```text
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
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;
```

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

```text
1: HELLO
2: WORLD
3: AZION
4: 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
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;
```

A function responde com os chunks que coletou:

```json
{"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
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;
```

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

```json
{"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
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;
```

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:

```json
{"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:

| API                    | Status                    |
| ---------------------- | ------------------------- |
| `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](/pt-br/documentacao/devtools/runtime/node.md): O status de cada módulo do Node.js no Azion Runtime, incluindo `stream`.
- [Use APIs do Node.js com polyfills](/pt-br/documentacao/guias/desenvolvimento-de-aplicacoes/functions-e-runtime/use-polyfills.md): Como o build transforma imports `node:` em código que é executado no Azion Runtime.
- [ReadableStream](/pt-br/documentacao/devtools/runtime/api-reference/readable-stream.md): A alternativa da Web Streams API para ler dados em chunks.
- [Documentação de stream do Node.js](https://nodejs.org/api/stream.html): A referência completa do Node.js para cada API de `node:stream` da tabela.