RxJS takeWhile, но включать последнее значение

Ответ 1

Там уже открыт PR, который добавляет необязательный inclusive параметр в takeWhile: https://github.com/ReactiveX/rxjs/pull/4115

Есть как минимум два возможных обходных пути:

1. используя concatMap():

   of('red', 'blue', 'green', 'orange').pipe(
     concatMap(color => {
       if (color === 'green') {
         return of(color, null);
       }
       return of(color);
     }),
     takeWhile(color => color),
   )
   

2. Использование multicast():

   of('red', 'blue', 'green', 'orange').pipe(
     multicast(
       () => new ReplaySubject(1),
       subject => subject.pipe(
         takeWhile((c) => c !== 'green'),
         concat(subject.take(1),
       )
     ),
   )
   

Я также использовал этот оператор, поэтому я добавил его в свой собственный набор дополнительных операторов RxJS 5: https://github.com/martinsik/rxjs-extra#takewhileinclusive

Этот оператор также обсуждался в этой проблеме RxJS 5: https://github.com/ReactiveX/rxjs/issues/2420

Январь 2019: обновлено для RxJS 6

Ответ 2

Если ваше сравнение таково, что вы точно знаете, что такое последний элемент (например, для !==), вы можете его повторно добавить:

Rx.Observable.from([2, 3, 4, 5, 6])
  .takeWhile((v) => v !== 4)
  .concat(Rx.Observable.of(4))
  .subscribe(console.log)

Ответ 3

Я столкнулся с одной и той же проблемой, мне нужен последний элемент для включения, поэтому я решил сохранить ссылку на подписку и отказаться от подписки в обратном вызове onNext, когда это условие было выполнено. Используя ваш пример кода, это будет:

const subscription = Observable.of('red', 'blue', 'green', 'orange')
  .subscribe(color => {
    // Do something with the value here
    if (color === 'green') {
      subscription.unsubscribe()
    }
  }) 

Это сработало для меня, потому что это также привело к тому, что наблюдаемый источник прекратил излучать, что мне нужно в моем сценарии. Я понимаю, что я не использую оператор takeWhile, но главная цель достигнута и без каких-либо обходных решений или дополнительного кода. Я не поклонник заставить вещи работать таким образом, чтобы они не были разработаны. Недостатки этого:

• Если есть какие-либо другие наблюдатели, подписанные, источник будет продолжать излучать.
• onCompleted не вызывается по какой-либо причине, если последний наблюдатель отменил подписку, но я проверил, что источник фактически прекращает выдавать.

Ответ 4

Вы можете использовать endWith(value) которое (в отличие от большого количества кода RxJS) очень хорошо самодокументируется.

const example = source.pipe(
                            takeWhile(val => val != 4), 
                            endWith(4));

PS. Также обратите внимание, что takeUntil не принимает предикат, поэтому, если вы пытались использовать этот оператор для решения этой проблемы, вы не сможете. Это совершенно другой метод подписи.

Официальные документы: https://rxjs-dev.firebaseapp.com/api/operators/endWith

https://stackblitz.com/edit/typescript-pvuawt

Ответ 5

В моем случае я не смог предсказать, каким будет окончательное значение. Я также просто хотел найти решение, включающее простые, простые операторы, и я хотел что-то, что я мог бы использовать повторно, поэтому я не мог полагаться на достоверность значений. Единственное, о чем я мог думать, это определить свой собственный оператор следующим образом:

import { pipe, from } from 'rxjs';
import { switchMap, takeWhile, filter, map } from 'rxjs/operators';

export function doWhile<T>(shouldContinue: (a: T) => boolean) {
  return pipe(
    switchMap((data: T) => from([
      { data, continue: true },
      { data, continue: shouldContinue(data), exclude: true }
    ])),
    takeWhile(message => message.continue),
    filter(message => !message.exclude),
    map(message => message.data)
  );
}

Это немного странно, но это работает для меня, и я могу импортировать и использовать его.