Rxjs как кэшировать switchMap только если новый не запущен?

Question

[Aetae]

Наговнокодил "в лоб" такую фигню:

function heavyRequest() {
  console.log('heavyRequest triggered');
  // здесь какой-то накладный запрос
  return timer(1000);
}

let subject$ = new Subject();
let pending = false;
let observable$ = subject$.pipe(
  tap(() => pending = true),
  switchMap((value) => { 
    // вызывается только один раз для каждого значения,
    // независимо от количества подписчиков
    return heavyRequest().pipe(mapTo(value))
  }),
  tap(() => pending = false),
  shareReplay(1),
  filter(() => !pending)
);

// 'first', затем 'second'
observable$.subscribe((v) => console.log('subscribe 1', v)); 

subject$.next('first');

setTimeout(() => {
  // 'first' сразу, затем 'second'
  observable$.subscribe((v) => console.log('subscribe 2', v));  
}, 2000);

setTimeout(() => {
  subject$.next('second');

  // только 'second'
  observable$.subscribe((v) => console.log('subscribe 3', v));  
}, 3000);

// Итоговый лог:
// "heavyRequest triggered"
// "subscribe 1" "first"
// "subscribe 2" "first"
// "heavyRequest triggered"
// "subscribe 1" "second"
// "subscribe 2" "second"
// "subscribe 3" "second"

Вопрос в том, как это сделать правильно по фенщую. Мозг уже плавится, но сильно подозреваю что всё должно быть намного проще.

Comments

[Антон Горецкий]

let pending$ = new BehaviorSubject(false);
let cachedValue$ = new BehaviorSubject(null);

let observable$ = subject$.pipe(
  withLatestFrom(pending$),
  filter(([value, pending]) => !pending),
  exhaustMap(([value]) => {
    pending$.next(true);
    return timer(1000).pipe(
      mapTo(value),
      tap(result => {
        pending$.next(false);
        cachedValue$.next(result);
      })
    );
  }),
  shareReplay(1)
);

[Aetae]

Антон Горецкий, что-то не сильно красивее.:)
Да и не работает, у тебя в итоге лог такой:

"heavyRequest triggered"
"subscribe 1" "first"
"subscribe 2" "first"
"heavyRequest triggered"
"subscribe 3" "first"
"subscribe 1" "second"
"subscribe 2" "second"
"subscribe 3" "second"

Т.е. subscribe 3 всё равно получает first, а не должен.

[Антон Горецкий]

у меня работает выводит: subscribe 1 first
subscribe 2 first
subscribe 3 first
subscribe 1 second
subscribe 2 second
subscribe 3 second
Версия rxjx https://unpkg.com/rxjs@7.3.0/dist/bundles/rxjs.umd...

[Aetae]

Антон Горецкий, обновил коммент(и вопрос чутка для наглядности).

[WbICHA]

Так вот как в поток новые данные подгружать... С этим эр-иксом все мозги уже переломал что там да как (хотя я не то, чтобы глубоко копал пока), спасибо, Aetae.)