RxJs - 从订阅Observable调用下一个主题

Question

我是Rx的新手，并且使用Http处理一些令牌管理工作。我的想法是，我跟踪一个令牌，并将此令牌附加到所有出站http请求的标头。

代码看起来像这样：

const request = (name, token, time) => { 
    console.log('attempt ' + name + ' with token ' + token); 
    return Rx.Observable.timer(time) 
     .do(() => { 
     console.log('sent ' + name + ' with token ' + token); 
     }) 
     .mapTo({ newToken: 'baz' }) // token baz arrives 
     .do(r => { 
     console.log('response for ' + name + ' received'); 
     }); 
}; 

const token$ = new Rx.ReplaySubject(1); 

token$.switchMap(token => request('request1', token, 0)) 
    //.do(rs => { token$.next(rs.newToken); }) // fails, causes a loop 
    .subscribe(
     // (rs) => { token$.next(rs.newToken); } // also fails with loop 
    ); 

token$.switchMap(token => request('request2', token, 1000)) 
    .subscribe(); 

// emit a token 

token$.next('bar'); 

// but how do I ensure 'request2' is sent with token 'baz' ? 

我的问题是，我怎么可以更新ReplaySubject与baz从请求返回观察到的？

我打过电话next从do在request1像这样：

token.switchMap(token => request('request1', token, 0)) 
    .do(rs => { token.next(rs.newToken); }) 
    .subscribe(); 

但这将它变成一个循环和request1重复使用令牌baz。在订阅的Observable中是否可以在do这样的主题上调用next？我想知道它是否违反了某些Rx合同或其他内容。

我自己也尝试request1这样的：

token.switchMap(token => request('request1', token, 0)) 
    .subscribe(rs => { token.next(rs.newToken); }); 

但是，同样的循环发生。可能是因为我打电话给next我是switchMap -ing！

任何想法如何能解决这个问题？谢谢！

jsbin here

Answer 1

好吧，如果我理解你的权利这map功能实际上是应该做一些外部调用和返回的令牌，那就应该被重复使用？而return { newToken: 'baz' };只是一些嘲讽的结果。

问题是，token.next('bar')不仅会为您的replaySubject添加新令牌，还会触发新的请求。

你可能应该做的是从tokenSubject拆分newRequestSubject。这样，您就可以推出新的令牌，而不会触发新的请求，因为如您已经注意到的那样，最终会导致无限循环。

然后可以使用newRequestSubject.withLatestFrom(tokenSubject).switchMap(request).subscribe()准备requestSubject并调用newRequestSubject.next(200, 'request1')推请求。

您甚至可以将tokenSubject上的replaySubject替换为behaviorSubject，以便您可以在创建时将初始标记传递给它。您可以拨打tokenSubject.next()来更新令牌。

作为一个侧面提示，它可能会更加优雅地消除所有的主题，例如mergeScan，它允许您从先前的请求中获取结果以将其馈送给下一个请求，但这可能是有点更难以把握。

newRequestSubject.mergeScan(request, initialToken); 

（也有是在RxJs没有switchScan，所以我们将失去的switch有用的行为。）