Буфер rxjs на основе предиката

Я хочу сделать что-то, что я считать должен быть довольно простым, но правильные операторы rxjs ускользают от меня.

Я хочу буферизовать наблюдаемую строку, очищая буфер, когда наблюдаемая создает совпадающую строку. Что-то вроде этого:

obs.pipe(magicBuffer(x => x === 'matching string'))

... для создания наблюдаемого массива строк.

Все операторы буфера, которые я могу найти, похоже, буферизуются на основе второго наблюдаемого, а не исходного источника.

Если мне нужно будет написать оператор, я это сделаю. Было бы очень просто написать. Но вместо этого я использую канонический подход к этой проблеме, если он есть.

— 06.02.2019 20:32

rxjs observable buffer

06.02.2019 20:31

Освоение Observables и Subjects в Rxjs:

Давайте начнем с основ и постепенно перейдем к более продвинутым концепциям в RxJS в Angular

Promise v/s Observable в Angular

В системах Push производитель определяет, когда отправить данные потребителю. Потребитель не знает, когда он получит эти данные.

Подсказка RxJS [filter, skipWhile]

Эта подсказка описывает разницу между операторами filter и skipWhile из библиотеки RxJS .

В чем разница между Promise и Observable?

Разберитесь в этом вопросе, и вы значительно повысите уровень своей компетенции.

123

Ответы 1

Вы можете использовать multicast для разделения исходного Observable, а затем просто buffer, пока он не выдаст совпадающую строку.

Это разделит исходные выбросы по характеру 'x'.

import { of, Subject } from 'rxjs'; 
import { filter, buffer, multicast } from 'rxjs/operators';

const source = of('a', 'b', 'c', 'x', 'd', 'x', 'e' , 'f', 'g', 'h', 'i', 'j', 'k', 'x', 'l', 'm');

source.pipe(
  multicast(
    () => new Subject(),
    s => s.pipe(
      filter(v => v !== 'x'),
      buffer(s.pipe(filter(v => v === 'x'))),
    )
  )
).subscribe(x => console.info(x));

Живая демонстрация: https://stackblitz.com/edit/rxjs-jad7ez

06.02.2019 21:26