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

An example with zip

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

// Borrow a book
const bookId$ = new Subject<number>();

// Refresh library card
const libraryCardUpdater$ = new BehaviorSubject<void>();

// re-emit book2 on card update to allow zip to pair card and book correctly
const book$ = combineLatest([bookId$, libraryCardUpdater$])
.pipe(
  switchMap(([bookId, _]) => fetchBook(bookId)),
  shareReplay(1)
);

const libraryCard$ = combineLatest([bookId$, libraryCardUpdater$])
.pipe(
  map(([bookId, _]) => fetchCard(bookId)),
  shareReplay(1)
);

book$.zip(libraryCard$)
.pipe(
  map(([book,libraryCard]) => `book: ${book.title}| borrower:${libraryCard.borrowerName}`),
)
.subscribe(console.log)

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

Output

book: Book1 | borrower: Borrower of Book1  <-- OK
book: Book2 | borrower: Borrower of Book2  <-- OK
book: Book2 | borrower: Borrower of Book2  <-- OK, but repeat due to libraryCardUpdater$
book: Book3 | borrower: Borrower of Book3  <-- OK

Answer 3

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

Answer 4

Rather than combineLatest() (which emits when any observable emits a value), I would use zip() which emits when all observables emit their value.

Also, I've updated some of your code to make things more declarative/reactive. While I know they were placeholders for an API call, there's no need for those functions when you can switchMap() from your bookId Subject.

const bookId = new Subject<number>();

const book$ = bookId.pipe(
  switchMap((bookId) =>
    timer(200).pipe(mapTo({ bookId, title: `Book${bookId}` }))
  ),
  shareReplay(1)
);

const libraryCard$ = bookId.pipe(
  switchMap((bookId) =>
    timer(300).pipe(mapTo({ borrowerName: `Borrower of Book${bookId}` }))
  ),
  shareReplay(1)
);

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

bookId.next(1);
setTimeout(() => bookId.next(2), 500);
setTimeout(() => bookId.next(2), 1000); // library card refresh
setTimeout(() => bookId$.next(3), 1500);

Answer 5

Here's my take on this:

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

const result$ = bookId$.pipe(
  switchMap(() => {
    const book$ = concat(of(null), fetchBook(bookId));

    const libraryCard$ = concat(
      of(null),
      concat(of(null), libraryCardUpdater$).pipe(
        concatMap(() => fetchLibraryCard(bookId))
      )
    );

    return combineLatest([book$, libraryCard$]);
  })
);

This will guarantee the following:

• whenever bookId$ changes, we start by emitting only once as expected: [null, null]
• then, there's a race between fetchBook and fetchLibraryCard calls. So we could either end up with [someBook, null] or [null, someLibraryCard]. At some point they should both be there and there will be another emission with [someBook, someLibraryCard]
• in the case where we've got for example [someBook, null] and the bookId changes, we'll then emit straight away [null, null]. Same for all the cases where the bookId changes
• as for the libraryCardUpdater$, it'll only ever trigger fetchLibraryCard and nothing else

Hopefully that matches what you wanted, always happy to edit my answer if I've missed anything

Answer 6

Sequential (not simultaneous) fetches make things easier:

const libraryCard$ = combineLatest([book$, libraryCardUpdater$]).pipe(
    switchMap(([book]) => concat(of({ borrowerName: <string>undefined }), fetchLibraryCard(book.bookId))),
    shareReplay(1)
);

The only thing that I need then is to add zero debouncer:

combineLatest([book$, libraryCard$]).pipe(debounceTime(0)).subscribe(...