How to wait for a consistent state in RxJS?

Question

I have a code to fetch book and library card associated with it:

// mimic http requests
const fetchBook = (bookId: number) => {
    const title = 'Book' + bookId;
    return timer(200).pipe(mapTo({ bookId, title }));
}
const fetchLibraryCard = (bookId: number) => {
    const borrowerName = 'Borrower of Book' + bookId;
    return timer(300).pipe(mapTo({ borrowerName }));
}

const bookId$ = new Subject<number>();

const book$ = bookId$.pipe(
    switchMap(bookId => fetchBook(bookId)),
    shareReplay(1)
);

// e.g. 'Refresh library card' button
const libraryCardUpdater$ = new BehaviorSubject<void>(undefined);

const libraryCard$ = combineLatest([bookId$, libraryCardUpdater$]).pipe(
    switchMap(([bookId]) => fetchLibraryCard(bookId)),
    shareReplay(1)
);

combineLatest([book$, libraryCard$]).subscribe(([book, libraryCard]) => {
    console.log('book:', book.title, '| borrower:', libraryCard.borrowerName)
})

bookId$.next(1);
setTimeout(() => bookId$.next(2), 500);
setTimeout(() => libraryCardUpdater$.next(), 1000);
setTimeout(() => bookId$.next(3), 1500);

The problem that I get inconsistent state in subscriber:

book: Book1 | borrower: Borrower of Book1  <-- OK
book: Book2 | borrower: Borrower of Book1  <-- Not OK
book: Book2 | borrower: Borrower of Book2  <-- OK
book: Book2 | borrower: Borrower of Book2  <-- OK, but redundant
book: Book3 | borrower: Borrower of Book2  <-- Not OK
book: Book3 | borrower: Borrower of Book3  <-- OK

I think about something like pushing undefined to libraryCard$ at the same moment when bookId$ is changed.

But how to do that in a reactive manner?

Update:

Library card should be always consistent with fetched book (or be undefined at loading time). bookId$ can be changed by user action at any time. Also library card can be updated at any time manually by user (libraryCardUpdater$). libraryCardUpdater$ emitting should re-fetch card, but shouldn't re-fetch book

Update2: I just realized that library card can be fetched sequentially after book. It is acceptable, although not perfect solution for end-user.

Answer 1

Testing your code in https://thinkrx.io/rxjs/ gives

where the last row is the same as your console.logs.

Changing to withLatestFrom instead of combineLatest removes unsynchronized book/card (#2 - 1/2/ & #5 - 2/3)

This is the code, with changes

• labels abbreviated
• timings divided by 10
• added id to cardUpdater$ and used Subject() with explicit .next() at start (cosmetic - still works with original BehaviorSubject).

const { rxObserver } = require('api/v0.3');
const rx = require('rxjs');
const { timer } = rx;
const { switchMap, map, mapTo, combineLatest, withLatestFrom, shareReplay } 
  = require('rxjs/operators');

// mimic http requests
const fetchBook = (bookId) => {
 return timer(20).pipe(mapTo({ bookId, title: 'b' + bookId }));
}
const fetchLibraryCard = (bookId) => {
  return timer(30).pipe(mapTo({ name: `c${bookId}` }));
}

const bookId$ = new rx.Subject();

const book$ = bookId$.pipe(
  switchMap(bookId => fetchBook(bookId)),
  shareReplay(1)
);

// e.g. 'Refresh library card' button
const cardUpdater$ = new rx.Subject();

const libraryCard$ = bookId$.combineLatest(cardUpdater$)
.pipe(
  switchMap(([bookId, cardId]) => fetchLibraryCard(bookId)),
  shareReplay(1)
)

const combined$ = libraryCard$.withLatestFrom(book$)
.pipe(
  map(([card,book]) => `b${book.title[1]}|c${card.name[1]}`),
)

// Marbles
bookId$.subscribe(rxObserver('id'))
book$.map(book=>book.title).subscribe(rxObserver('book'))
cardUpdater$.subscribe(rxObserver('card update'))
libraryCard$.map(card=>card.name).subscribe(rxObserver('libraryCard$'))
combined$.subscribe(rxObserver('combined'))

// Events
bookId$.next(1);
cardUpdater$.next(1)
setTimeout(() => bookId$.next(2), 50);
setTimeout(() => cardUpdater$.next(2), 100);
setTimeout(() => bookId$.next(3), 150);

---

One thing that puzzles me is this emit you want to remove.

book: Book2 | borrower: Borrower of Book2  <-- OK, but redundant

It's triggered by cardUpdater$ event, can be removed with distinctUntilChanged() in combined$, but doing so makes the card refresh superfluous.

It feels like you want a cardId which changes on card refresh, and re-issues the same book on the new card.

Something like this has a more orthogonal feel

const { rxObserver } = require('api/v0.3');
const rx = require('rxjs');
const { timer } = rx;
const { switchMap, map, mapTo, combineLatest, withLatestFrom, shareReplay } 
  = require('rxjs/operators');

const fetchBook = (bookId) => {
 return timer(20).pipe(mapTo({ bookId, title: 'b' + bookId }));
}
const fetchLibraryCard = (cardId) => {
  return timer(30).pipe(mapTo({ name: `c${cardId}` }));
}

const bookId$ = new rx.Subject();
const book$ = bookId$.pipe(
  switchMap(bookId => fetchBook(bookId)),
  shareReplay(1)
);

const cardUpdater$ = new rx.Subject();
const card$ = cardUpdater$.pipe(
  switchMap(cardId => fetchLibraryCard(cardId)),
  shareReplay(1)
);

const issue$ = book$.merge(card$).pipe(
  switchMap(() => card$.withLatestFrom(book$)),
  map(([card,book]) => `${book.title}|${card.name}`),
)

// Marbles
bookId$.subscribe(rxObserver('id'))
book$.map(book=>book.title).subscribe(rxObserver('book'))
cardUpdater$.subscribe(rxObserver('card update'))
card$.map(card=>card.name).subscribe(rxObserver('libraryCard$'))
issue$.subscribe(rxObserver('combined'))

// Events
bookId$.next(1);
cardUpdater$.next(1)
setTimeout(() => bookId$.next(2), 50);
setTimeout(() => cardUpdater$.next(2), 100);
setTimeout(() => bookId$.next(3), 150);

Answer 2

You have to turn things around. Your source of truth need to be the bookId$, and from that constructued observable you can get the book and libraryCard:

const bookId$ = new ReplaySubject<number>(1);

const libraryCardUpdater$ = new Subject<void>();

const libraryCardBook$ = combineLatest([
  bookId$.pipe(
    distinctUntilChanged(),
    switchMap(bookId => fetchBook(bookId))
  ),
  libraryCardUpdater$.pipe(
    switchMap(() => this.bookId$),
    switchMap((bookId) => fetchLibraryCard(bookId))
  )
]).pipe(
  map(([ book, libraryCard ]) => ({ book, libraryCard })),
  startWith({ book: undefined, libraryCard: undefined }),
  shareReplay(1)
);

const book$ = libraryCardBook$.pipe(map(({ book }) => book);
const libraryCard$ = libraryCardBook$.pipe(map(({ libraryCard }) => libraryCard);