RxJs будут последними в случае завершения, в противном случае вставьте задержку между значениями

Question

Я дам полное объяснение, но в основном мне нужно убедиться, что

1. существует некоторая задержка между излучаемыми значениями
2. если наблюдаемое завершено, просто верните последнее значение.

ДАЛИ

У меня есть две наблюдаемые, каждая из которых испускает 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$ значение должно игнорироваться.

   • подсказка: если $ запрос завершен , также cacheAndRequest $ завершается во время подписки.

2. , если $request НЕ завершено во время подписки, немедленно отправьте значение из $ cache и значение из $ request как минимум через 500 мс. Другими словами, между этими двумя значениями должна быть задержка.

Вот детская площадка:

Скопируйте этот код здесь

Answer 1

Чтобы добиться этого, я немного изменил ваши cache$ и request$, чтобы идентифицировать наблюдаемое:

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)),
  );

См. мой рабочий пример с Stackblitz .