Mais Streams

No post anterior falei sobre os 3 tipos de streams do NodeJS: Readable, Writable e Duplex. Neste post, vamos mergulhar um pouco mais fundo nessa API para você entender como ela funciona na prática e em situações do dia a dia.

PassThrough

O PassThrough é uma implementação bem simples de uma stream Transform que apenas repassa os dados de entrada diretamente para a saída, sem alterá-los. Um caso de uso muito comum para ele é a injeção de logs no meio de um pipeline de dados.

import { stdout } from "node:process";
import { PassThrough, Readable } from "node:stream";
import { styleText } from "node:util";

const data = ["10\n", "22\n", "31\n", "14\n", "523\n"];
const readable = Readable.from(data);

const loggerStream = new PassThrough();
const consoleStream = new PassThrough();
const otherStream = new PassThrough();

loggerStream.on("data", (chunk) => {
    console.log(styleText(["cyan"], `Logger: ${chunk.toString()}`));
});

consoleStream.on("data", (chunk) => {
    console.log(styleText(["green"], `Console: ${chunk.toString()}`));
});

otherStream.on("data", (chunk) => {
    console.log(styleText(["yellow"], `Other: ${chunk.toString()}`));
});

readable.pipe(loggerStream).pipe(consoleStream).pipe(otherStream).pipe(stdout);

Na teoria o PassThrough não deve alterar os dados, apesar de ser possível, já que é uma stream Transform.

Async Iterators e Async Generators

A API de streams do node é compatível com async iterators e async generators. O que pode tornar o código mais fácil de ler ou ser escrito em alguns caso.

Para isso usamos a função pipeline que cria um fluxo de streams, async iterators ou async generators.

import { stdout } from "node:process";
import { PassThrough, Readable } from "node:stream";
import { pipeline } from "node:stream/promises";

// Função geradora assíncrona
async function* sumGenerator(stream) {
  // o for await...of é usado para iterar sobre os chunks de dados do stream
  for await (const chunk of stream) {
    const data = Number(chunk.toString().trim());

    const sumData = data + 100;

    // O yield é usado para enviar o resultado para o próximo estágio do pipeline
    // O valor deve ser uma string ou Buffer
    yield sumData.toString() + "\n";
  }
}

const data = [" 10 ", " 20 ", " 30 ", " 40 ", " 50 "];
const readable = Readable.from(data);

try {
    await pipeline(
      readable,
      sumGenerator,
      stdout
    );
} catch (error) {
    console.error("Pipeline failed.", error);
}

Requisições HTTP

Você pode pensar que as streams são algo que você não usaria no dia a dia. Mas, na realidade algumas APIs usam streams e você acaba não percebendo.

O módulo HTTP do NodeJS é um exemplo disso. Podemos criar um servidor http usando o node:http e o objeto request, que representa a requisição do usuário, será um stream Readable e o objeto response, que representa a resposta do servidor, será uma stream Writable.

Isso acontece porque a requisição viaja pela rede e chega ao nosso servidor como um fluxo contínuo de dados (uma stream). Graças a isso, o servidor não precisa esperar o carregamento de 100% do body da requisição para começar a processá-la. O mesmo princípio de performance se aplica na hora de devolver a resposta ao cliente.

import { createServer } from "node:http";
import { PassThrough } from "node:stream";
import { styleText } from "node:util";

const logger = new PassThrough();

logger.on("data", (chunk) => {
    const prefix = styleText(["cyan"], "Logger: \n");
    const message = styleText(["yellow", "bold"], chunk.toString());
    console.log(prefix + message);
});

// request é um Readable
// response é um Writable
const server = createServer((request, response) => {
  // redirecionando o que chega no request body para o response
  request.pipe(logger).pipe(response);
});

server.listen(3000, () => {
  console.log("Server is listening on port 3000");
});

No express podemos adicionar um middleware para lidar com requisições json usando app.use(express.json()) e podemos devolver uma resposta em json usando response.json().

import express from "express";

const app = express();

app.use(express.json());

app.post("/", async (request, response) => {
    const body = request.body;
    response.json({ message: "Dados recebidos com sucesso", data: body });
});

O módulo node:http não tem essa abstração, mas podemos construir essa abstração agora que entendemos melhor as streams do node:http

import { createServer } from "node:http";

// Middleware para extrair o corpo da requisição como JSON
async function json(request, response) {
  let body = "";

  // como o request é uma stream, consumimos ela com o for await of
  for await (const chunk of request) {
    // acumulamos os pedaços do corpo da requisição
    body += chunk;
  }

  try {
    // transformamos o corpo da requisição em JSON e adicionamos ao objeto request
    body = JSON.parse(body);
  } catch {
    // Se não for um json válido, definimos o corpo como um objeto vazio
    body = {};
  }

  // adicionamos o corpo da requisição ao objeto request
  request.body = body;
}

// Middleware para adicionar o método status ao objeto response
async function statusResponse(request, response) {
  response.status = (statusCode) => {
    response.statusCode = statusCode;
    return response;
  }
}

// Middleware para enviar respostas JSON
async function jsonResponse(request, response) {
  response.json = async (data) => {
    const status = response.statusCode || 200;

    response.writeHead(status, { "Content-Type": "application/json" });
    response.end(JSON.stringify(data));
  }
}

const middlewares = [json, statusResponse, jsonResponse];

// Mudamos para uma função assíncrona para lidar com os middlewares
const server = createServer(async (request, response) => {
  const { method, url } = request;

  // Chamamos cada middleware em sequência, passando o request e response
  for (const middleware of middlewares) {
    await middleware(request, response);
  }

  if (method === "POST" && url === "/") {
    const body = request.body || {};

    response.status(201).json({ message: "Dados recebidos com sucesso", data: body });
  }
});

server.listen(3000, () => {
  console.log("Server is listening on port 3000");
});

Conexão Serial

Outra biblioteca que usa streams por baixo dos panos é a serialport, uma biblioteca para fazer a conexão com dispositivos através da porta serial.

Você pode se conectar a um Arduino ou ESP32, por exemplo, para coletar os dados que eles enviam e salvar em um arquivo de log. A conexão que a serialport faz é usando uma stream Duplex.

import { ReadlineParser } from "@serialport/parser-readline";
import { SerialPort } from "serialport";
import { createWriteStream } from "node:fs";
import { pipeline } from "node:stream/promises";
import { Transform } from "node:stream";

const baudRate = 38400; // taxa de transmissão em bits por segundo do dispositivo serial
const path = "COM10"; // porta serial do dispositivo (ex: COM1, /dev/ttyUSB0, etc.)

// uma stream Duplex para ler e escrever dados na porta serial
const port = new SerialPort({ path, baudRate });

// um parser para ler os dados da porta serial linha por linha
const parser = new ReadlineParser({ delimiter: "\n" });

// uma stream para colocar os dados linha a linha
const lineStream = new Transform({
    transform(chunk, encoding, callback) {
        const line = chunk.toString().trim();
        if (line) {
            this.push(line + "\n");
        }
        callback();
    }
})

// uma stream para escrever os dados recebidos da porta serial em um arquivo de log
const logFileStream = createWriteStream("serial_log.txt");


try {
    await pipeline(port, parser, lineStream, logFileStream);
} catch (error) {
    console.error("Erro ao processar os dados da porta serial:", error);
}

Conclusão

Muitas vezes, quando estudamos os conceitos de Streams no Node.js, tudo parece muito teórico ou voltado apenas para a leitura de arquivos gigantescos em disco. Mas, como vimos, as streams estão por trás de muitas ferramentas que usamos diariamente.

Elas estão presentes quando levantamos um servidor HTTP, quando o Express faz o parse de um JSON no body da requisição e até quando nos comunicamos com hardwares externos usando portas seriais.

Entender como as streams funcionam por debaixo dos panos te dá uma vantagem imensa: você passa a escrever códigos mais performáticos, que consomem muito menos memória e conseguem processar informações sob demanda.