组合运算符的反应模式，其中一个运算符依赖于另一个运算符

Question

我有一个使用 REST 服务的应用程序。每次有人例如，它都会发出请求。单击一个按钮。但是，它需要先获取令牌，然后每 20 分钟刷新一次。这是代表流的所需组合的弹珠图...

TOKEN SOURCE   ---------A--------------B--------------C----------------D-
REQUEST SOURCE ----1-----------2--3---------4-----------5-------6--------


RESULT SOURCE  ---------A1-----A2-A3--------B4----------C5------C6-------

通常情况下，请求源会触发与令牌源的最新值相结合的结果，但是当令牌流上没有发出任何项目时会出现异常 - 在这种情况下，请求会被缓冲直到第一个令牌到达然后发送。

combineLatest operator is almost there but it triggers when either stream emits. The marble diagram for sample 运算符似乎也很接近，但它根据我想要的时间间隔限制输出。

哪个 operator/chain 运算符适用于此实例？

我正在使用 RxJS 制作原型，但我需要在 RxSwift 中实现。

Answer 1

您已接近解决方案。重要的细节是 Source (2) 投射它的所有元素，而 Source (1) 不直接投射，并且当只有 Source (1) 发生变化时，拥有这些额外的信息可以帮助我们丢弃。

s1.combineLatest(s2, (a, b) => [a, b])
  .distinctUntilChanged(v => v[1])

这是测试设置：

var s1 = Rx.Observable.interval(997).take(10).map(i => String.fromCharCode(65 + i))
var s2 = Rx.Observable.interval(251).take(10).delay(500).skip(1)

s1.combineLatest(s2, (a, b) => [a, b])
  .distinctUntilChanged(v => v[1])
  .subscribe(value => console.log(value))

输出：

["A",1] ["A",2] ["A",3] ["A",4] ["B",5] ["B",6] ["B",7] ["B",8] ["C",9]

Answer 2

由于您希望仅根据您的 REQUEST SOURCE 触发您的数据发射，这可能是最好的方式来启动您的流然后切换 - 或者在这种情况下 flatMap - 在 TOKEN SOURCE 上并将这两者结合起来 - 不需要高级运算符：

（注意：我的假设是，您的 TOKEN SOURCE 确实重放了最新的令牌：

var token$ = Rx.Observable.interval(1000).take(10).map(() => Math.random()).publishReplay(1); // simulating the token source: new token every second and replaying the latest token
var requests$ = Rx.Observable.interval(250).take(10); // simulating the request source: emitting a new request every 250ms
token$.connect();

requests$
    .flatMap(request => token$
        .take(1)
        .map(token => ({request, token}))
    )
    .subscribe(function(requestWithToken) {
        console.log("Matched: ", requestWithToken);
    });

您可以在此处查看直播 fiddle：https://jsfiddle.net/rn8rzufg/

组合运算符的反应模式，其中一个运算符依赖于另一个运算符

Reactive pattern for combing operator where one is dependent on another

operators

reactive-programming

rxjs

rx-swift