如何在 RxJS 中等待一致状态？

Question

我有一个代码来获取与之相关的书籍和借书卡：

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

我在订阅者中得到不一致状态的问题：

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

我想到了一些事情，比如在发生变化的同时undefined推动。libraryCard$bookId$

但如何以被动的方式做到这一点呢？

更新：

借书证应始终与所取图书一致（或在undefined加载时）。bookId$可以随时通过用户操作进行更改。用户还可以随时手动更新借书卡（libraryCardUpdater$）。libraryCardUpdater$发出应该重新获取卡，但不应该重新获取书本

更新2： 我刚刚意识到图书卡可以在书后依次获取。对于最终用户来说，虽然不是完美的解决方案，但这是可以接受的。

Answer 1

在https://thinkrx.io/rxjs/中测试您的代码给出

其中最后一行与您的 console.logs 相同。

更改为withLatestFrom而不是combineLatest删除不同步的书籍/卡片 (#2 - 1/2/& #5 - 2/3)

这是代码，经过修改

• 标签缩写
• 时间除以10
• 在开始时添加了 id并与显式cardUpdater$一起使用（修饰 - 仍然适用于原始BehaviorSubject）。Subject().next()

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

让我困惑的一件事是你想要删除这个发射。

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

它由事件触发，可以使用incardUpdater$删除，但这样做会使卡片刷新变得多余。distinctUntilChanged()combined$

感觉就像您想要cardId在卡刷新时进行更改，并在新卡上重新发行同一本书。

像这样的东西有一种更正交的感觉

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

Answer 2

一个例子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);

输出

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

顺序（非同时）获取使事情变得更容易：

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

我唯一需要的就是添加零防抖器：

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