Operators - ganhando poderes para tratar streams de dados com RxJS

Operators ou Operadores são os poderes que você ganha para tratar streams de dados.

Não é muito interessante você poder criar um fluxo de dados e pode receber esse fluxo em algum lugar, se você não pode fazer nada entre esses dois pontos.

Sendo assim é aqui que os operators entram, eles são quase como “filtros" que podemos encadear no processo para fazer algum tipo de tratamento nos dados que estamos fazendo o consumo, fazendo com que um código assíncrono e complexo possa ser composto por funções de forma declarativa.

Você tem uma coleção de operadores fornecidas pela lib RxJS ou desenvolvidas conforme necessidade, os operadores são úteis para tratar o dado fornecido pelo fluxo.

Para construção de um Operator utilizamos do paradigma funcional, usando o conceito de funções puras.

Operadores são funções, nada mais e nada menos do que funções.

import { of, map } from 'rxjs';

of(1, 2, 3)
  .pipe(map((x) => x * x)) // o map é nosso operator nesse fluxo.
  .subscribe((v) => console.log(`value: ${v}`));

Agora que arranhamos o entendimento sobre operators vamos falar de fato sobre eles.

Existem dois tipos de operators

Pipeables Operators: são pipes que podem ser canalizados no sentido de algo que possa passar por um filtro, eles não alteram a instancia do Observable que esta sendo executado, em vez disso eles retornam um novo observable onde a lógica de inscrição e baseada no primeiro Observable.

const observable = new Observable(function subscribe(subscriber) {
  const id = setInterval(() => {
    subscriber.next(1)
        subscriber.next(2)
        subscriber.next(3)
  }, 1000);
});

observable
    .pipe(map(v) => v * v) // A funcao executada dentro do pipe, no caso o map e um Pipeable Operator
    .subscribe((v) => console.log(`value: ${v}`));

    // value: 1
    // value: 4
    // value: 9 

Um pipeable operator e uma função pura que recebe um Observable como entrada e retorna outro Observable, E exatamente uma função pura pois o Observable anterior que fica nao sofre qualquer modificação.

Os pipeables operators tambem podem ser executadas de forma encadeada atraves do metodo pipe ja visto anteriormente.

observable
    .pipe(
        operator1(),
        operator2(),
        operator3(),
    )
    .subscribe((v) => console.log(`value: ${v}`));

Creation Operators Operadores de criação são funções autonomas que podem ser chamadas para criar um Observable com algum determinado comportamento predefinido ou combinando outros Observables.

Exemplo:

import { of, map } from 'rxjs';

of("a", "b", "c") // of gera um Observable que emite um valor após o outro
.pipe(map((v) => `${v}1`))
.subscribe((v) => console.log(value: ${v}));

// value: a1
// value: b1
// value: c1

High Order Observables

Um Observable pode emitir valores comuns como strings e números, mas é muito provável que você em algum momento precise manipular um Observable de outro Observable, eles são chamados de High Order Observables, um conceito parecido com High Order Functions, vamos analisar o seguinte cenário.

Temos um Observable que faz a leitura de um determinado diretório para obter todos os arquivos de texto.

const fs = require('fs');
const path = require('path');
const { Observable } = require('rxjs')

function readDirectory(pathDir) {
  return new Observable(subscriber => {
    try {
      let files = fs.readdirSync(pathDir)
      files = files.forEach(file => {
        subscriber.next(path.join(pathDir, file))
      })
      subscriber.complete()
    } catch (error) {
      subscriber.error(error)
    }
  })
}

Ok, já temos um Observable.

Agora temos os arquivos, como poderíamos ler o conteúdo desses arquivos em forma de stream com algumas modificações? Criando um outro Observable, mas nesse caso um Pipeable Operator.

function createPipeableOperator(operatorFN) {
  return function name(source) {
    return Observable.create(subscriber => {
      const sub = operatorFN(subscriber)
      source.subscribe({
        next: sub.next,
        error: sub.error || (e => subscriber.error(e)),
        complete: sub.complete || (v => subscriber.complete()),
      })
    })
  }
}

A partir da função createPipeableOperator podemos criar novas funções como Observables que podem ter algum papel na hora de ler os dados do diretório, algo como ler o arquivo apenas de uma determinada extensão.

// ler o arquivo.
function readFile() {
  return createPipeableOperator(subscriber => ({
    next(pathFile) {
      try {
        const content = fs.readFileSync(pathFile, { encoding: 'utf-8' })
        subscriber.next(content.toString());
      } catch (error) {
        subscriber.error(error)
      }
    }
  }))
}

//ler com uma determinada extensão.
function getFilesWithExtension(extension) {
  return createPipeableOperator(subscriber => ({
    next(text) {
      if (text.endsWith(extension)) {
        subscriber.next(text)
      }
    }
  }))
}

No fim para executar esse programa temos algo nessas estrutura.

readDirectory(./caminho_do_diretorio)
  .pipe(
    getFilesWithExtension('.txt'),
    readFile(),
  )
  .subscribe(console.log)

Com isso conseguiremos ler os arquivos de um determinado diretório com a extensão .txt usando Observables de Observables, parece complicado, mas tem a ver mais com prática do que com qualquer outra coisa.

Categoria dos Operadores

Agora você consegue entender a gama de possibilidades que você pode fazer usando os operadores, pois existe operadores para os mais variados propósitos divididos em várias categorias filters, joins, multicasting etc…

segue o link da documentação oficial para que você consiga explorar mais informações dos operators