Problema com a compreensão de RXJS Observable Output

Question

Eu sou bastante novo em rxjs e estou tentando aprender.

Eu estava lendo este artigo médio e tentando entender esta saída RXJS:

import {Observable} from 'rxjs-es';
let output = Observable.interval(500)
             .map(i => [1,2,3,4,5,6][i]);

let result = output.map(num1 => num1)
    .filter(num1 => num1 > 4)
    .reduce((num1, num2) => num1 + num2);

result.subscribe(number => console.log(number));

Produto: 27

Não entendo como a saída acabou sendo 27 e como essa redução está funcionando (e quais são esses dois argumentos para o redutor).

Alguém pode me esclarecer o que está acontecendo? (Eu tentei executá-lo no codesandbox, mas ele está gerando um erro quando executo este código)

Answer 1

Entendimento Filtere Reduceoperadores

Como outros mencionaram nos comentários, o artigo ao qual você se refere está usando uma versão mais antiga do rxjs que possui uma sintaxe diferente. Para este exemplo, usarei a sintaxe mais recente que começou na versão 6.

Em rxjs, existem vários operadores disponíveis para transformar os valores emitidos por meio de um fluxo. Normalmente, são importados como:

import { filter, reduce } from 'rxjs';

Existem também muitas funções geradoras disponíveis para criar um fluxo de valores. intervalé uma dessas funções que criará um fluxo que emite inteiros sequenciais a cada nmilissegundos. Importado como:

import { interval } from 'rxjs/operators';

Vamos criar um fluxo simples:

number$ = interval(1000); // emit number every 1 second
// output: 0, 1, 2, 3, 4, 5...

Podemos aplicar operadores a este fluxo para transformar as emissões:

O uso de filteré bastante simples. Ele simplesmente emite valores que passam no teste de verdade fornecido (exatamente como o Array.filter()método).

numbersLessThan4$ = numbers$.pipe(
    filter(number => number < 4)
);
// output: 0, 1, 2, 3

O reduceoperador é um pouco mais complexo e se comporta exatamente como o Array.reduce()método. Uma função é aplicada a cada valor emitido e é capaz de armazenar um valor que pode ser referenciado quando a próxima emissão for avaliada.

reduceleva dois parâmetros. A primeira é uma função que recebe a emissão atual ( cur) e o resultado acumulado anterior ( acc) e retorna um novo valor acumulado. O segundo é um valor inicial para acc.

exemplo:

sumOfNumbers$ = numbers$.pipe(
    reduce((acc, cur) => acc + cur, 0)
);

Então, vamos ver o que reduceacontece quando numbers$emite os 3 primeiros números:

• 0
  • cur recebe o valor de emissão atual 0
  • acc começa com o padrão fornecido 0
  • a expressão acc + curretorna0
• 1
  • cur recebe o valor de emissão atual 1
  • acc recebe o valor retornado anteriormente 0
  • a expressão acc + curretorna1
• 2
  • cur recebe o valor de emissão atual 2
  • acc recebe o valor retornado anteriormente 1
  • a expressão acc + curretorna3

Então isso é legal. Podemos obter bastante lógica em uma linha simples de código. Uma coisa importante reduceé que ele não será emitido até que a fonte observável seja concluída. Atualmente, numbers$nunca é concluído ( interval()emite inteiros sequenciais indefinidamente).

Podemos usar o take()operador para completar o fluxo depois que um certo número de valores são emitidos.

Exemplo:

numbers$ = interval(1000).pipe(take(5)); // completes after 5 emissions

sumOfNumbers$ = numbers$.pipe(
    // receives 5 values (0, 1, 2, 3, 4) and performs the logic described above.
    reduce((acc, cur) => acc + cur, 0) 
);
// output: 10

Vários operadores podem ser usados ​​para transformar as emissões. Basta fornecer vários dentro de pipe():

sumOfNumbersLessThan4$ = numbers$.pipe(
    filter(number => number < 4),
    reduce((acc, cur) => acc + cur, 0) 
);
// output: 6