Streams no Node

[jemluz]

O que são Streams

Streams permitem que o usuário acesse/consuma conteúdos extensos/pesados, sem ter que aguardar ele ser baixado por completo primeiro. Na realidade, o conteúdo é fornecido em pequenas "parcelas" e essas são baixadas a medida que o usuário necessita.

Por exemplo em um filme, o primeiro download é realizado nos primeiros minutos, e a medida que o usuário vai assistindo esses minutos, downloads de partes seguintes vão ocorrendo.

O exemplo acima é um Writables Stream / Streams de escrita. Utilizadas para enviar conteúdo aos poucos. Também temos os Readable Streams / Streams de leitura. Que é o que o node pode fazer ao processar um .cvs gigantesco e ir lendo os dados do .csv aos poucos, na medida em que ele pode ir registrando esses dados em um banco.

[jemluz]

Portas

No NodeJS, todas as portas são streams.

Como por exemplo: Request e Response.

[jemluz]

Streams de leitura, escrita e transformação.

Leitura

O exemplo abaixo é uma stream de leitura que realiza uma iteração a cada 1s, e transmite/retorna - com this.push() - o resultado enquanto ele não chega no número 100. A função _read() é utilizada para o processamento da leitura. Para encaminhar o valor ele precisa converter para um Buffer - com Buffer.from(). Toda stream de leitura precisa exterder da classe Readable.

import { Readable } from 'node:stream'

class OneToHundredStream extends Readable {
  index = 1

  _read() {
    const i = this.index++

    setTimeout(() => {
      if (i > 100) {
        this.push(null)
      } else {
        const buf = Buffer.from(String(i))

        this.push(buf)
      }
    }, 1000);
  }
}

Transformação

Toda stream de transformação é feita para intermediar uma stream de leitura com uma stream de escrita. A função _transform() é utilizada para o processamento da transformação/tratamento. O parâmetro chunck se refere a informação que foi transmitida pela stream anterior (uma stream de leitura). O parâmetro callback irá entregar a informação tratada/processada para a próxima stream (uma stream de escrita).

class InverseNumberStream extends Transform {
  _transform(chunk, encoding, callback) {
    const transformed = Number(chunk.toString()) * -1

    callback(null, Buffer.from(String(transformed)))
  }
}

Escrita

A stream de escrita pode ser vista como o fim da linha para o caminho percorrido pela informação.

class MultiplyByTenStream extends Writable {
  _write(chunk, encoding, callback) {
    console.log(Number(chunk.toString()) * 10)
    callback()
  }
}

[jemluz]

Como as Streams se conectam com o servidor HTTP do node

Todas as portas de entrada e saída no node são streams. Isso quer dizer que os objetos request e response são streams.

Request é uma readable stream. E o Response é uma writable stream.

Utilizamos o método fetch para realizar uma requisição para a url da porta 3334 (que é aberta por outro arquivo) que envia uma stream de leitura (ela vai dentro da req)

class OneToHundredStream extends Readable {
  index = 1

  _read() {
    const i = this.index++

    setTimeout(() => {
      if (i > 100) {
        this.push(null)
      } else {
        const buf = Buffer.from(String(i))

        this.push(buf)
      }
    }, 1000);
  }
}
// Abaixo temos a base de como o node funciona. Veja esse metodo fetch como uma porta aberta não se fecha e por ela nossos dados podem trafegar indeterminadamente.
// POST ou PUT utilizamos para enviar informação
// GET utilizamos para receber informação
fetch('http://localhost:3334', {
  method: 'POST',
  body: new OneToHundredStream(),
})

Aqui abrimos um servidor na porta 3334 e capturamos a stream de leitura do código acima dentro do objeto req. Em seguida encadeamos com a stream de transformação e encaminhamos o resultado para o objeto res.

import http from 'node:http'
import { Transform } from 'node:stream'

class InverseNumberStream extends Transform {
  _transform(chunk, encoding, callback) {
    const transformed = Number(chunk.toString()) * -1

    console.log(transformed)

    callback(null, Buffer.from(String(transformed)))
  }
}

// req => ReadableStream
// res => WritableStream

const server = http.createServer((req, res) => {
  return req
    .pipe(new InverseNumberStream())
    .pipe(res)
})

server.listen(3334)

O método pipe() encaminha os dados de uma stream para outra. Como se fosse um encanamento cada stream e os dados fossem o fluxo de água.

[jemluz]

Consumindo uma stream completa

Podemos usar o for await para aguardar o consumo completo dos dados de uma stream antes de executar o restante do código.

Isso é útil para trabalhar com JSON por exemplo, já que você não consegue tratar objetos antes de tê-los por completo.

const server = http.createServer(async (req, res) => {
  const buffers = []

  for await (const chunk of req) {
    buffers.push(chunk)
  }

  const fullStreamContent = Buffer.concat(buffers).toString()

  console.log(fullStreamContent)

  return res.end(fullStreamContent)
})

[jemluz]

Entendendo Buffers

Buffers são representações hexadecimais de dados complexos (como acentos e caracteres especiais por exemplo).

No JS não existe um jeito nativo de processar esses dados, então o buffers surge justamente para resolver esse problema. São estruturas de dados que representam informações binárias.

const buf = Buffer.from("hello")

console.log(buf.toJSON())

[jemluz]

O que djabo é um middleware

São interceptadores. Isto é, são funções intermediadoras das nossas requisições. São muito úteis para tratamento de dados ou error handling por exemplo.

Middlewares sempre receberão req e res como parâmetros.

[jemluz]

Acessando um objeto dinâmicamente

No caso de precisar passar um nome de propriedade como parâmetro de função, para que ele seja acessado por um objeto dentro da função, utilizamos o []

select(table) {
    const data = this.#database[table] ?? []

    return data
  }