RxJS parte 2: operators, do map à search-as-you-type

Introdução

Na parte 1 a gente usou map, takeUntil e switchMap no cronômetro, e pediu para aceitar isso sem olhar por dentro. O problema é que, sem entender o que é um operator, o pipe vira uma caixa preta, e também fica difícil montar uma search-as-you-type sem disparar uma enxurrada de buscas a cada tecla. Neste artigo a gente vai ver o que é um operator, os que usamos no dia a dia, e no final construir esse campo que filtra os resultados conforme a pessoa escreve.

Os exemplos usam RxJS 7 (7.8.x). Todos os trechos de código deste artigo foram executados, e as saídas mostradas são as saídas reais.

O que é um operator

interval e fromEvent são funções de criação, e criam observables do nada. Já map, filter e takeUntil são operators, e recebem um observable e devolvem outro observable. Um operator, chamado de "pipeable operator" na documentação, é uma função com esta cara:

(input: Observable<A>) => Observable<B>

Operators são como os map e filter de array, e também são diferentes porque trabalham valor a valor conforme os valores chegam, e não sobre uma lista pronta. Para ver que não tem mágica, vamos escrever um operator na mão, só com o que a gente viu na parte 1:

import { Observable, of } from "rxjs";

// An operator: takes an observable and returns a new observable
function double() {
  return (source$) =>
    new Observable((subscriber) => {
      const subscription = source$.subscribe({
        next: (value) => subscriber.next(value * 2),
        error: (error) => subscriber.error(error),
        complete: () => subscriber.complete(),
      });
      return () => subscription.unsubscribe(); // teardown: forward cancellation
    });
}

of(1, 2, 3)
  .pipe(double())
  .subscribe({
    next: (v) => console.log("next:", v),
    complete: () => console.log("complete"),
  });

next: 2
next: 4
next: 6
complete

O double não mexe no observable de entrada, e cria um novo que, quando alguém faz subscribe, faz subscribe na source e repassa cada valor transformado. É assim que todo operator do RxJS funciona por baixo dos panos, com mais cuidado com casos de borda, claro, e por isso o pipe() retorna um novo observable em vez de alterar o original, como a gente viu na parte 1.

O of é uma função de criação que emite os valores que você passar, de forma síncrona, e completa. Vamos usar bastante para testar operators.

Na prática a gente não precisa escrever o double, porque map((v) => v * 2) já faz isso. Quando escrevemos .pipe(a(), b(), c()), estamos montando uma cadeia em que cada operator faz subscribe no anterior.

of e from

Antes dos operators, duas funções de criação que facilitam os experimentos:

import { of, from } from "rxjs";

of(10, 20, 30).subscribe((v) => console.log("of:", v));
from([1, 2]).subscribe((v) => console.log("from array:", v));
from(Promise.resolve("ok")).subscribe((v) => console.log("from promise:", v));
console.log("end of synchronous code");

of: 10
of: 20
of: 30
from array: 1
from array: 2
end of synchronous code
from promise: ok

• of(a, b, c) emite cada argumento.
• from(x) converte arrays, Promises, iteráveis e outros "observable-likes" num observable. Como a gente viu na parte 1, a Promise continua assíncrona.

map e filter

Se a gente já usou Array.prototype.map e filter, já sabe 90% destes dois. A diferença é que eles trabalham valor a valor, conforme os valores chegam, e não sobre uma lista pronta.

import { of, map, filter } from "rxjs";

of(1, 2, 3, 4, 5, 6)
  .pipe(
    filter((n) => n % 2 === 0), // keep only the even numbers
    map((n) => n * 10) // transform each value
  )
  .subscribe((v) => console.log(v));

20
40
60

Em marble:

source:         -1-2-3-4-5-6-|
filter(even):   ---2---4---6-|
map(n => n*10): ---20--40--60|

tap

Às vezes a gente quer fazer alguma coisa ao lado do stream: um console.log para debugar, salvar algo no localStorage, e/ou atualizar um contador. Para isso existe o tap, e ele recebe os mesmos callbacks de um observer e repassa o valor sem alterar:

import { of, map, tap } from "rxjs";

of("a", "b")
  .pipe(
    tap((v) => console.log("before map:", v)),
    map((v) => v.toUpperCase()),
    tap({
      next: (v) => console.log("after map:", v),
      complete: () => console.log("tap saw complete"),
    })
  )
  .subscribe((v) => console.log("subscriber received:", v));

before map: a
after map: A
subscriber received: A
before map: b
after map: B
subscriber received: B
tap saw complete

Cada valor atravessa a cadeia inteira antes de o próximo começar, e não é "primeiro todos passam pelo map, depois todos passam pelo tap", como aconteceria com arrays.

Regra de bolso: map transforma, tap faz efeito colateral. O valor que você retorna de dentro de um tap é ignorado, e se a gente queria trocar o valor, era um map. E sem subscribe nada roda, nem o tap.

take e takeUntil

Muitos observables nunca terminam, como interval e cliques, e os operators da família take decidem quando parar. Ao parar, eles completam o stream e fazem unsubscribe da source, e aí o teardown roda e nada vaza.

take(n) deixa passar os n primeiros valores e completa:

import { interval, take } from "rxjs";

interval(100)
  .pipe(take(3))
  .subscribe({
    next: (n) => console.log("tick", n),
    complete: () => console.log("complete: interval was shut down"),
  });

tick 0
tick 1
tick 2
complete: interval was shut down

takeUntil(notifier$) deixa passar valores até o notifier emitir, e foi ele que parou o cronômetro na parte 1. Aqui o notifier é um timer(350), que emite uma vez depois de 350 ms:

import { interval, timer, takeUntil } from "rxjs";

interval(100)
  .pipe(takeUntil(timer(350)))
  .subscribe({
    next: (n) => console.log("tick", n),
    complete: () => console.log("complete: timer stopped it"),
  });

tick 0
tick 1
tick 2
complete: timer stopped it

interval(100):          --0-1-2-3-4-...
timer(350):             -------(t|)
takeUntil(timer(350)):  --0-1-2|
                        (cada - ≈ 50 ms)

Também existe o takeWhile(condition), que para quando a condição fica falsa, e na prática take e takeUntil resolvem a maioria dos casos.

scan e reduce

O reduce do array junta todos os itens num valor só. O RxJS também tem um reduce, mas ele só emite quando a source completa, e num observable de cliques, que nunca completa, ele nunca emitiria nada. O scan é a versão ao vivo: ele acumula e emite o acumulado a cada valor.

import { of, scan, reduce } from "rxjs";

const purchases$ = of(10, 25, 5);

purchases$.pipe(scan((total, value) => total + value, 0)).subscribe((t) => console.log("scan:", t));

purchases$.pipe(reduce((total, value) => total + value, 0)).subscribe((t) => console.log("reduce:", t));

scan: 10
scan: 35
scan: 40
reduce: 40

source:  -10-25-5-|
scan:    -10-35-40|
reduce:  ---------(40|)

O scan é a base de qualquer "state que muda com o tempo" em RxJS, e na parte 4 a gente vai construir um store inteiro com ele.

debounceTime

Quando alguém digita "rxjs" num campo de busca, o evento input dispara quatro vezes: r, rx, rxj, rxjs. Se cada um disparasse uma busca, seriam quatro buscas, e três delas inúteis.

O debounceTime(ms) segura cada valor e só deixa passar quando a source fica ms milissegundos em silêncio. Se chegar um valor novo antes disso, o anterior é descartado e a contagem recomeça.

Para simular alguém digitando, vamos criar um observable que emite em momentos definidos:

import { Observable, debounceTime } from "rxjs";

// Emits each [ms, value] at the given time and completes at the end
const emitAt = (...events) =>
  new Observable((subscriber) => {
    const ids = events.map(([ms, value]) => setTimeout(() => subscriber.next(value), ms));
    const end = setTimeout(() => subscriber.complete(), Math.max(...events.map(([ms]) => ms)) + 1);
    return () => [...ids, end].forEach(clearTimeout);
  });

const typing$ = emitAt([0, "r"], [100, "rx"], [200, "rxj"], [300, "rxjs"], [800, "rxjs o"], [900, "rxjs op"]);

const start = Date.now();
typing$.pipe(debounceTime(300)).subscribe({
  next: (term) => console.log(`~${Math.round((Date.now() - start) / 100) * 100}ms: search "${term}"`),
  complete: () => console.log("complete"),
});

~600ms: search "rxjs"
~900ms: search "rxjs op"
complete

typing$:            abcd----e(f|)
debounceTime(300):  ------d--(f|)

cada caractere ≈ 100 ms; a = "r", b = "rx", c = "rxj", d = "rxjs", e = "rxjs o", f = "rxjs op"
(f|) = f e o complete no mesmo instante

De seis eventos, só duas buscas, e dois detalhes importam:

• O "rxjs" saiu em ~600 ms: 300 ms depois da última tecla.
• O "rxjs op" saiu "antes da hora", em ~900 ms, porque a source completou. Quando a source completa, o debounceTime entrega o valor pendente na hora em vez de jogá-lo fora.

Primo próximo: o throttleTime(ms) faz o contrário, deixa passar o primeiro valor e ignora os seguintes durante ms. É bom para eventos como scroll e resize, e para um campo de busca o debounceTime costuma ser o certo.

distinctUntilChanged

O distinctUntilChanged() ignora um valor se ele for igual ao anterior (comparação com ===):

import { of, distinctUntilChanged } from "rxjs";

of("rx", "rx", "rxjs", "rxjs", "rx").pipe(distinctUntilChanged()).subscribe((v) => console.log(v));

rx
rxjs
rx

Ele compara só com o imediatamente anterior, e o último "rx" passou porque veio depois de "rxjs". Se você quer "nunca repetir nada", existe o distinct(), mas cuidado: ele guarda todos os valores já vistos para sempre.

Para objetos, === compara referências, então dois objetos iguais "no conteúdo" são diferentes, e você pode passar um comparador:

import { of, distinctUntilChanged } from "rxjs";

of({ id: 1, name: "Ana" }, { id: 1, name: "Ana" }, { id: 2, name: "Bia" })
  .pipe(distinctUntilChanged((previous, current) => previous.id === current.id))
  .subscribe((u) => console.log(u.name));

Ana
Bia

startWith

O startWith(value) emite um valor antes de qualquer valor da source, e é útil para a tela não começar vazia:

import { of, startWith } from "rxjs";

of("rx", "rxjs").pipe(startWith("")).subscribe((v) => console.log(JSON.stringify(v)));

""
"rx"
"rxjs"

A ordem dos operators importa

Como cada operator recebe a saída do anterior, trocar dois de lugar pode mudar tudo:

import { of, filter, take } from "rxjs";

const numbers$ = of(1, 2, 3, 4, 5, 6);

numbers$.pipe(filter((n) => n % 2 === 0), take(2)).subscribe((n) => console.log("filter → take:", n));
numbers$.pipe(take(2), filter((n) => n % 2 === 0)).subscribe((n) => console.log("take → filter:", n));

filter → take: 2
filter → take: 4
take → filter: 2

Nesse exemplo, duas coisas podem acontecer: a gente pega os dois primeiros pares, ou pega os dois primeiros números (1 e 2) e só depois filtra. Leia o pipe de cima para baixo, como uma linha de montagem.

Criando operators com a função pipe

Na parte 1 a gente viu que pipe é um método do observable. Ele também existe como função solta, importada de "rxjs", e serve para juntar vários operators num só, reutilizável:

import { of, pipe, map, filter } from "rxjs";

// a new operator, built from existing operators
const normalizeTerm = pipe(
  map((text) => text.trim().toLowerCase()),
  filter((text) => text.length >= 2)
);

of("  RxJS ", "a", " Map").pipe(normalizeTerm).subscribe((v) => console.log(JSON.stringify(v)));

"rxjs"
"map"

O normalizeTerm é exatamente aquela função Observable → Observable do começo do artigo, e a gente vai usá-lo no projeto.

Projeto: search-as-you-type

Vamos juntar o que vimos e construir um campo que filtra uma lista conforme a pessoa digita. Para não depender de servidor nenhum (isso fica para a parte 3), a lista vai ser local, e vai ser justamente a dos operators do RxJS.

O comportamento que queremos:

1. Normalizar o texto (sem espaços nas pontas, minúsculo).
2. Esperar a pessoa parar de digitar por 300 ms.
3. Não refazer a busca se o termo não mudou (ex.: digitou um espaço e apagou).
4. Mostrar a lista completa enquanto o campo está vazio.

Crie o projeto como na parte 1 (npm create vite@latest search -- --template vanilla-ts e npm install rxjs), ou use o StackBlitz. O HTML:

<body>
  <h1>RxJS operators</h1>
  <input id="search" placeholder="Search operator..." autocomplete="off" />
  <p id="status"></p>
  <ul id="results"></ul>
  <script type="module" src="/src/search.ts"></script>
</body>

E o src/search.ts:

import { fromEvent, map, debounceTime, distinctUntilChanged, startWith, tap, pipe } from "rxjs";

const OPERATORS = [
  "map", "filter", "tap", "take", "takeUntil", "takeWhile", "scan", "reduce",
  "debounceTime", "throttleTime", "distinctUntilChanged", "startWith",
  "switchMap", "mergeMap", "concatMap", "exhaustMap", "catchError", "retry",
  "share", "shareReplay",
];

const input = document.querySelector<HTMLInputElement>("#search")!;
const status = document.querySelector<HTMLElement>("#status")!;
const list = document.querySelector<HTMLUListElement>("#results")!;

const normalize = pipe(map((text: string) => text.trim().toLowerCase()));

const term$ = fromEvent(input, "input").pipe(
  map(() => input.value), // from the event to the typed text
  normalize,
  debounceTime(300), // wait 300 ms of silence
  startWith(""), // full list before any typing
  distinctUntilChanged(), // don't repeat the search if the term didn't change
  tap((term) => console.log(`searching "${term}"`))
);

const results$ = term$.pipe(
  map((term) => OPERATORS.filter((op) => op.toLowerCase().includes(term)))
);

results$.subscribe((found) => {
  status.textContent = `${found.length} operator(s)`;
  list.innerHTML = "";
  for (const name of found) {
    const item = document.createElement("li");
    item.textContent = name;
    list.append(item);
  }
});

Nesse exemplo, duas coisas podem acontecer se a gente errar a ordem dos operators: a lista começa vazia, e/ou um espaço no final dispara uma busca inútil. Vale olhar cada pedaço do que o código faz.

map(() => input.value): o fromEvent emite o Event, mas o que interessa é o texto. Ler input.value na hora do evento é a forma mais simples, e o valor já vem como string no TypeScript, e (e.target as HTMLInputElement).value também funciona.

normalize: um operator feito com a função pipe. Aqui ele só tem um map, mas é o lugar certo para crescer, remover acentos, por exemplo, sem bagunçar a cadeia principal.

debounceTime(300): quem digita "switch" rápido gera uma busca só, e não seis.

startWith("") antes do distinctUntilChanged(): começamos com o termo vazio, para a lista aparecer inteira logo de cara, e a ordem aqui importa. Se a pessoa digitar algo e apagar tudo, o termo volta a ser "", que é diferente do anterior, então a lista completa volta. Mas se ela digitar "switch" e depois "switch " (com um espaço), o normalize transforma os dois em "switch" e o distinctUntilChanged barra a segunda busca.

tap: só para a gente ver no console quando uma busca realmente acontece.

Um subscribe só, no final, como no cronômetro, e todo o resto é composição.

Simulando uma pessoa que digita "s", "sw", "swi", "switch" rapidinho e, depois de uma pausa, coloca um espaço no final, o console mostra:

searching ""
searching "switch"

Duas buscas: a inicial e a do "switch". O espaço no final não gerou busca nova, e na tela o switchMap aparece sozinho na lista com o status "1 operator(s)".

normalized:            -a-b-c-d---------d-------
debounceTime(300):     -----------d---------d---
startWith(""):         x----------d---------d---
distinctUntilChanged:  x----------d-------------

x = "", a = "s", b = "sw", c = "swi", d = "switch" (o "switch " com espaço vira "switch" no normalize)

Exercício: troque o includes por uma busca que ignore acentos e destaque o trecho encontrado. Dica: dá para resolver tudo dentro do normalize e do map final, sem mexer no resto.

Resumo

• Um operator é uma função Observable → Observable. O pipe() encadeia operators e cada um faz subscribe no anterior.
• of e from criam observables a partir de valores, arrays e Promises, e são ótimos para testar.
• map transforma, filter seleciona, tap faz efeito colateral sem alterar o valor.
• take(n) e takeUntil(notifier$) completam o stream e fazem unsubscribe da source.
• scan acumula e emite a cada valor; reduce só emite quando a source completa.
• debounceTime espera um silêncio; distinctUntilChanged ignora repetições seguidas; startWith emite um valor inicial.
• A ordem dos operators muda o resultado. A função pipe() cria operators reutilizáveis.

E se a lista estivesse num servidor? Aí cada termo vira um HTTP request, que demora, pode falhar e também pode chegar fora de ordem. Um map não dá conta, e a gente precisa transformar cada termo num observable de resposta. Esse é o trabalho do switchMap e dos seus irmãos, o assunto da parte 3.

Links para leitura:

• Documentação oficial: Operators
• Árvore de decisão de operators (Operator Decision Tree)
• Learn RxJS: Operators
• RxMarbles