RxJS:如何不订阅初始值和/或未定义

RxJS: How to not subscribe to initial value and/or undefined?

本文关键字：未定义 何不订 RxJS      更新时间：2023-09-26

作为RxJS的新手，我经常创建一个主题，该主题在未来具有值，但最初是undefined。它只能是第一次undefined。我目前使用filter来跳过undefined值，但这很麻烦，因为我在任何地方都这样做，因为我只需要一次。（也许我在这里做错了什么？我可以在mySubject通过onNext获得第一个值后以某种方式订阅它吗？

var mySubject = new Rx.BehaviorSubject(undefined);
mySubject.filter(function(value) {
  return value !== undefined;
}).subscribe(function(value) {
  // do something with the value
});

使用 new Rx.ReplaySubject(1) 而不是 BehaviorSubject 。

如 如 是否应该能够使用 skip 运算符跳过第一个值： 

var mySubject = new Rx.BehaviorSubject(undefined);
mySubject.pipe(skip(1)).subscribe(function(value) {
  // do something with the value
});

mySubject.pipe( skipWhile( v => !v ) );

现在我正在使用 filter 运算符，但我不知道这是否是一个很好的解决方案：

var mySubject = new Rx.BehaviorSubject().filter(x => !!x);
mySubject.subscribe(value => { /* will receive value from below */);
mySubject.next('value');
mySubject.subscribe(value => { /* also receives the value */ });

我发现这在RxJS和RxSwift中都令人沮丧。（想要一个值主体，并能够等待第一个值）。

对于JS，我目前只是在主题中隐藏一个过滤版本，如下所示：

    let mySubject = new Rx.BehaviorSubject();
    mySubject.wait = mySubject.pipe(filter(v=>v!==undefined));

因此，主题仍会公开以供发布，但客户端不必重复筛选器。

    mySubject.wait.subscribe((v)=>{...});

有时需要 behaviorSubject，其中初始值无关紧要，并且在流中工作时异步需要当前值，在我们的例子中，多个链承诺是在处理或从流中的任何位置获取数据时通过用户取消来处理的。

这可以使用以下方法实现。

// for user related commands
this.commandSource = new BehaviorSubject(CONTINUE);
// filtering over initial value which is continue to make it as a different pipe
const stopPipe = commandSource.pipe(filter(val => val === STOP));
const fetchStream = Observable.fromPromise(this.fetchDetails);
merge(fetchStream, stopPipe).pipe(
 take(1),
 takeWhile(() => commandSource.value === CONTINUE),
 concatMap((response) => {
  // fetch Another response you can return promise directly in concatMap
  // return array of response [1 ,2 ,3];
  return this.fetchYetAnotherDetails;
 }),
 // we can add this to stop stream in multiple places while processing the response
 takeWhile(() => commandSource.value === CONTINUE),
 // triggers parallelly values from the concatMap that is 1, 2 , 3
 mergeMap(() => // massage the response parallelly using )
 finalize(() => thi
  commandSource.complete())
).subscribe(res => {
 // handle each response 1, 2, 3 mapped
}, () => {
 // handle error
}, () => {
 // handle complete of the stream
});
// when user, clicks cancel, this should stop the stream.
commandSource.next(STOP)
相关文章：