Как объединить каждый внутренний наблюдаемый (Rx Js) с внешним и испускать один массив?

Question

Когда я запускаю этот код:

timer(1000).pipe(
  expand(() => of('a')),
  take(3)
)
  .subscribe(data => alert(data));

Я получаю предупреждение 3 раза: предупреждение с 0, предупреждение с 'a', предупреждение с 'a'. Я хотел бы получить только одно предупреждение с, скажем, array [0, a, a]. Как совместить все значения?

Answer 1

Ваш вопрос интересен и (как отмечает Майкл Д. в своем ответе) очень конкретный c вариант использования.

Пример следующего кода, работающего в StackBlitz

Вместо использования scan (который даст вам текущее накопленное значение для каждого из 3 выбросов) я бы предложил уменьшить , который сработает только один раз, когда исходный код завершен.

import { of, timer } from 'rxjs';
import { reduce, expand, take } from 'rxjs/operators';

timer(1000).pipe(
  expand(() => of('a')),
  take(3),
  reduce((acc, val) => { acc.push(val); return acc; }, [])
).subscribe(val => console.log('Output:', val));

// Вывод: [0, a, a]

Answer 2

Это похоже на очень конкретный вариант использования c, и есть несколько способов сделать это.

Вариант 1: scan и filter операторы

timer(1000).pipe(
  expand(() => of('a')),
  take(3),
  scan((acc, curr) => {
    acc.push(curr);    // <-- push each notification to the array
    return acc;        
  }, []),              // <-- empty array `[]` as seed value 
  filter(values => values.length === 3)    // <-- emit only if notification length is 3
).subscribe(data => alert(data));

Обратите внимание, что наблюдаемый оператор using scan отправляет 3 уведомления: [0], [0, 'a'] и [0, 'a', 'a']. Мы явно ограничиваем отправку только последнего уведомления, потому что знаем, что оно должно иметь длину 3 (из-за take(3)).

Вариант 2: scan и takeLast операторы

timer(1000).pipe(
  expand(() => of('a')),
  take(3),
  scan((acc, curr) => {
    acc.push(curr);    // <-- push each notification to the array
    return acc;        
  }, []),              // <-- empty array `[]` as seed value 
  takeLast(1)          // <-- emit the last 1 notification
).subscribe(data => alert(data));

Вариант 3: scan и last операторы

timer(1000).pipe(
  expand(() => of('a')),
  take(3),
  scan((acc, curr) => {
    acc.push(curr);    // <-- push each notification to the array
    return acc;        
  }, []),              // <-- empty array `[]` as seed value 
  last()               // <-- emit the last notification
).subscribe(data => alert(data));

last оператор предлагает еще одну опцию, аналогичную оператору filter. Вы можете передать предикат, и уведомление будет отправлено только в том случае, если оно удовлетворяет этому предикату.

timer(1000).pipe(
  expand(() => of('a')),
  take(3),
  scan((acc, curr) => {
    acc.push(curr);    // <-- push each notification to the array
    return acc;        
  }, []),              // <-- empty array `[]` as seed value 
  last((values) => values.includes('a'))     // <-- emit only if array contains 'a'
).subscribe(data => alert(data));

Вариант 4: reduce оператор

См. Другое ответ от Натана Бека.