javascript - RxJs 如果完成则取最后，否则在值之间插入延迟

Question

我会给出完整的解释，但基本上我需要确保

1. 发出的值之间有一些延迟
2. 如果 observable 完成，只返回最新的值。

给于

我有两个可观察对象，每个都发出 0 或 1 个值，然后完成:

 //1. lazy observable that emit 0 or two value created at time of subscription and then completes
 const cache$ = this.getFromCacheLazy();

 //2.  ReplaySubject, that emits value in 1-1500ms and completes
 const request$ = this.executeRequest(); 

我将它结合到可观察的，首先发出缓存值(如果有的话)然后请求值:

 const cacheAndRequest$ = concat(cache$, request$);

一段时间后我订阅了它:

 setTimeout(() => {
     cacheAndRequest$.Subscribe(console.log);
 }, someDelay)

我该怎么做

修改或管道cacheAndRequest$，以便:

1. 如果 $request 在订阅时已经完成，cache$ 值必须被忽略。

   • 提示:如果 $request 已完成，则 cacheAndRequest$ 在订阅时也已完成。

2. 如果 $request 在订阅时未完成，则立即从 $cache 发出值，并在最早 500 毫秒后从 $request 发出值。换句话说，这两个值之间必须存在延迟。

这里是 Playground :

复制this代码 here

Answer 1

为了实现这一点，我稍微修改了你的 cache$ 和 request$ 来识别 observable:

const cache$ = getFromCacheLazy().pipe(
  map(cache => {
    return {source: 'cache', value: cache};
  })
);

const request$ = executeRequest().pipe(
  map(request => {
    return {source: 'request', value: request};
  })
);

我像你一样留下了 concat。

然后我应用了 withLatestFrom 函数来获取最新的发射值。如果最新发出的值来自 request$，我将其返回，否则我连接可观察对象以返回 cache$ 值并等待至少 500 毫秒以发出 request$ :

const $subscribeDelayed = timer(subscriptionDelay)
  .pipe(
    withLatestFrom(cacheAndRequest$), // Provide the latest value from cacheAndRequest$
    switchMap(res => {
      // if request$ is already completed at the time of subscription, cache$ value is ignored
      // else cache$ is emitted and request$ is emitted after at least 500ms
      return res[1].source === 'request'
        ? of(res[1].value)
        : concat(
          of(res[1].value),
          request$.pipe(
            delay(500),
            map(res => res.value)
          )
        );
    }),
    tap(res => console.log(res)),
  );

参见 my working example on Stackblitz .