rxjs share with interval вызывает проблему при ожидании следующей итерации интервала

Я думаю, это должно сработать:

const newSubscription = new Subject();

const covidCases$ = interval(60 * 1000).pipe(
  takeUntil(newSubscription),
  repeat(),
  switchMap(() =>
    this.covidService.getCovidCases().pipe(
      /* ... */
    ),
  ),
  takeUntil(this.stopPolling),
  shareReplay(1),
  src$ => defer(() => (newSubscription.next(), src$))
);

Я заменил timer(1, 60 * 1000) + retry() на interval(60 * 1000).

Я полагал, что для перезапуска таймера (interval()) мы должны повторно подписаться на него. Но перед повторной подпиской мы должны сначала unsubscribed от нее.

Вот что делают эти строки:

interval(60 * 1000).pipe(
  takeUntil(newSubscription),
  repeat(),
  /* ... */
)

У нас есть таймер, пока не сработает newSubscription. Когда это произойдет, takeUntil выдаст complete notification, а затем отменит подписку на свой источник (в данном случае источник, созданный interval).

repeat перехватит это complete notification и повторно подпишется на наблюдаемый источник (source = interval().pipe(takeUntil())), что означает, что таймер перезапустится.

shareReplay(1) гарантирует, что новый подписчик получит последнее переданное значение.

Тогда очень важно разместить src$ => defer(() => (newSubscription.next(), src$)) после shareReplay. Используя defer(), мы можем определить момент прибытия нового подписчика.

Если бы вы поместили src$ => defer(() => (console.log('sub'), src$)) выше shareReplay(1), вы бы увидели, что sub выполнено, зарегистрировано только один раз после создания первого подписчика. Поместив его под shareReplay(1), вы должны видеть это сообщение в журнале каждый раз, когда создается подписчик.

Вернемся к нашему примеру, когда новый подписчик зарегистрирован, newSubscription будет излучать, что означает, что таймер будет перезапущен, но поскольку мы также используем repeat, полное уведомление не будет передано shareReplay, если только не отправит stopPolling.

StackBlitz demo.

Этот код создает наблюдаемый объект. Я думаю, что вам следует добавить Replaysubject вместо Observable.

Replaysubjects дает возможность генерировать то же событие при возникновении новой подписки.

timer(1, 60000).pipe(
  switchMap(() =>
    this.covidService.getCovidCases().pipe(
      tap(result => {
        if (!result.page.totalElements) {
          this.stopPolling.next();
        }
      }),

      map(data => {
        return data.cases;
      }),
      tap(results =>
        results.sort(
          (a, b) =>
            new Date(b.covidDateTime).getTime() -
            new Date(a.covidDateTime).getTime(),
        ),
      ),
    ),
  ),
  retry(),
  share(),
  takeUntil(this.stopPolling),
).subscribe((val)=>{this.covidcases.next(val)});

Эта модификация приводит к созданию таймера один раз, поэтому, когда вы подпишетесь на тему, он немедленно выдаст последнее значение.

Вы можете написать оператор, который подталкивает количество вновь добавленных подписчиков к заданной теме:

const { Subject, timer, Observable } = rxjs;
const { takeUntil, repeat, map, share } = rxjs.operators;

// Operator
function subscriberAdded (subscriberAdded$) {
  let subscriberAddedCounter = 0;
  return function (source$) {
    return new Observable(subscriber => {
      source$.subscribe(subscriber)
      subscriberAddedCounter += 1;
      subscriberAdded$.next(subscriberAddedCounter)
    });
  }
}

// Usage
const subscriberAdded$ = new Subject();

const covidCases$ = timer(1, 4000).pipe(
  takeUntil(subscriberAdded$),
  repeat(),
  map(() => 'testValue'),
  share(),
  subscriberAdded(subscriberAdded$)
)

covidCases$.subscribe(v => console.info('subscribe 1: ', v));
setTimeout(() => covidCases$.subscribe(v => console.info('subscribe 2: ', v)), 5000);
subscriberAdded$.subscribe(v => console.warn('subscriber added: ', v));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.3/rxjs.umd.min.js"></script>

Будущие возможности:

• Вы можете легко обновить оператора, чтобы уменьшить количество, если вы хотите отреагировать на отписавшихся

! важно

TakeUnit + repeat уже был опубликован @ AndreiGătej. Я только предоставил альтернативный способ получения события при добавлении подписчика.

Запуск stackblitz с машинописным текстом

Если оператору subscriberAdded требуется корректировка, дайте мне знать, и я обновлю