RxJS takeWhile 但包含最后一个值

Posted 2023-02-23

【中文标题】RxJS takeWhile 但包含最后一个值

【英文标题】：RxJS takeWhile but include the last value

【发布时间】：2017-11-22 07:12:42

【问题描述】：

我有一个像这样的 RxJS5 管道

Rx.Observable.from([2, 3, 4, 5, 6])
  .takeWhile((v) =>  v !== 4 )

我想保留订阅直到看到 4，但我希望最后一个元素 4 也包含在结果中。所以上面的例子应该是

2, 3, 4

但是，根据official document，takeWhile 运算符不包含在内。这意味着当它遇到与我们给出的谓词不匹配的元素时，它会立即完成流，而不是最后一个元素。结果，上面的代码实际上会输出

2, 3

所以我的问题是，我可以实现 takeWhile 并使用 RxJS 发出最后一个元素的最简单方法是什么？

【问题讨论】：

有点滑稽，但最简单的方法是.takeWhile(v => v < 5)

更认真：github.com/martinsik/rxjs-extra/blob/master/src/operator/…

为我工作github.com/MatthiasKunnen/rxjs-take-while-inclusive

可能没有帮助，但相关（这是我在找到这篇文章后最终使用的）：RxJava has a takeUntil method which takes a predicate。与 takeWhile 的不同之处在于它确实包含最后一项。

【参考方案1】：

从 RxJS 6.4.0 开始，现在可以使用 takeWhile(predicate, true)。

已经有一个打开的 PR 向takeWhile 添加了一个可选的inclusive 参数：https://github.com/ReactiveX/rxjs/pull/4115

至少有两种可能的解决方法：

使用concatMap():

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

使用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 issue 中讨论过：https://github.com/ReactiveX/rxjs/issues/2420

2019 年 1 月：针对 RxJS 6 更新

【参考方案2】：

如果您的比较结果是您确切知道最后一个元素是什么（例如!==），您可以自己重新添加：

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

【讨论】：

看起来这就是endWith 基本上是在做github.com/ReactiveX/rxjs/blob/…

（我意识到它可能是一年前的新东西）

【参考方案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

【讨论】：

注意 - 并非所有运营商都在 learn-rxjs.io 上，所以它目前不是最好的学习场所

直接回答问的问题，在哪里知道最终的值，我喜欢这个最简单直接的。当然，正如其他人所指出的，在 v6.4 中为运算符添加一个包含参数是理想的。

【参考方案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)
  );

这有点奇怪，但它对我有用，我可以导入并使用它。

【参考方案6】：

2019 年 3 月更新，rsjx 版本6.4.0：takeWhile 终于有了一个可选的inclusive 参数，允许保留第一个打破条件的元素。所以现在的解决方案是简单地将 true 作为takeWhile 的第二个参数传递：

import  takeWhile  from 'rxjs/operators';
import  from  from 'rxjs';

const cutOff = 4.5
const example = from([2, 3, 4, 5, 6])
.pipe(takeWhile(v => v < cutOff, true ))
const subscribe = example.subscribe(val =>
  console.log('inclusive:', val)
);

输出：

inclusive: 2
inclusive: 3
inclusive: 4
inclusive: 5

住在这里：

https://stackblitz.com/edit/typescript-m7zjkr?embed=1&file=index.ts

请注意，5 是第一个打破条件的元素。请注意，endWith 并不是真正的解决方案，因为您有像 v < cutOff 这样的动态条件并且您不知道最后一个元素是什么。

感谢@martin 指出此拉取请求的存在。