在 rxjs 中排队 http 调用

Question

我正在研究一个会话服务，用于检查身份验证令牌是否已过期。如果是，它将调用以刷新令牌。在此请求期间，所有传入的请求都应排队，并在请求完成后发出。之后，所有传入的请求都可以在没有队列的情况下通过，直到令牌再次过期。我为此绘制了一张弹珠图：

1. ---a---b---c--d-----e--
2. -t-------f------t------
3. ---a---b---------cd-e--

我将 1. 命名为incoming$Observable，2. 是valve$- 如果为 true，则请求可以通过，如果为 false，则应将它们排队。当它变为 true 时，将触发排队。

到目前为止我做了什么？我认为这应该通过添加一个名为receiver$的中间可观察量来完成，它根据valve$更改其值。当valve$为 true 时，它只返回一个简单的主题，如果为 false，则返回一个能够记录值的主题。

receiver$ = valve.pipe(
map((value) => {
if (value) {
return new Subject();
} else {
return (new Subject()).pipe(
shareReplay(),
);
}
})
);

然后，在incoming$中获得的每个新值都应添加到当前可观察的recevier$

：

incoming$.pipe(
combineLatest(receiver$),
).subscribe((incomingValue, recevier) => {
recevier.next(incomingValue);
});

这是我无法理解的部分。每当阀门变为真时，我都需要receiver$中的最后两个值。倒数第二个将保存队列，最后一个将保存活动主题。通过合并它们，我可以实现我的目标。我不知道如何实现这一点以及如何管理订阅。此外，对于这样一个看似简单的用例来说，这看起来过于复杂。

实现此行为的最佳方法是什么？

Answer 1

您可以通过仅使用基于值形式合并两个不同流的concatMap来执行此操作valve$.请注意，这要求valve$和incoming$都与share()共享。

valve$
.pipe(
concatMap(v => v
? incoming$.pipe(takeUntil(valve$))
: incoming$
.pipe(
takeUntil(valve$),
bufferCount(Number.POSITIVE_INFINITY),
mergeAll(),
)
),
)
.subscribe(console.log)

现场演示：https://stackblitz.com/edit/rxjs6-demo-d3bsxb?file=index.ts

Answer 2

您可以考虑以下解决方案。

首先，创建一个主题，通过该主题发出要发出的所有请求

const requests$ = new Subject<Observable<any>>()

然后你创建一个主题，通过它传达阀门的状态，即你是否可以立即执行请求，或者你必须缓冲它

const valve$ = new Subject<boolean>();

现在，您可以创建一个流，该流仅在阀门打开时才传递请求，即如果valve$发出的最后一个值是true

const openStream$ = valve$.pipe(
switchMap(valve => {
if (valve) {
return requests$;
} else {
return empty();
}
})
);

您还可以创建一个流，在阀门关闭时缓冲所有请求

const bufferedStream$ = requests$.pipe(
bufferToggle(valve$.pipe(filter(valve => !valve)), () => valve$.pipe(filter(valve => valve))),
mergeMap(bufferedCalls => bufferedCalls)
)

现在您所要做的就是mergeopenStream$并bufferedStream$并subscribe生成的流，如下所示

merge(openStream$, bufferedStream$).pipe(
mergeMap(request => request)
)
.subscribe(httpCallResult => {// do stuff})

我已经使用以下数据测试了此解决方案，使用字符串的可观察量模拟真实的 http 调用

const requests$ = new Subject<Observable<string>>();
setTimeout(() => {requests$.next(of('A'))}, 50);
setTimeout(() => {requests$.next(of('B'))}, 60);
setTimeout(() => {requests$.next(of('C'))}, 100);
setTimeout(() => {requests$.next(of('D'))}, 110);
setTimeout(() => {requests$.next(of('E'))}, 130);
setTimeout(() => {requests$.next(of('F'))}, 250);
setTimeout(() => {requests$.next(of('G'))}, 260);
setTimeout(() => {requests$.next(of('H'))}, 300);
setTimeout(() => {requests$.next(of('I'))}, 310);
setTimeout(() => {requests$.next(of('L'))}, 330);

const valve$ = new Subject<boolean>();
setTimeout(() => {valve$.next(true)}, 30);
setTimeout(() => {valve$.next(false)}, 80);
setTimeout(() => {valve$.next(true)}, 120);
setTimeout(() => {valve$.next(false)}, 200);
setTimeout(() => {valve$.next(true)}, 290);