问题描述
我有一个用例,我需要限制传出 http 请求的数量。是的,我确实在服务器端有速率限制器,但前端也需要对活动 http 请求的数量进行限制。出于这个原因,我试图实现一个滑动窗口协议,在任何时候我都会只有 n 个活动请求。
这种使用 Rxjs 的方法通常可以正常工作,请参见此处: https://jsbin.com/pacicubeci/1/edit?js,output
但我不清楚如何对 http 拦截器使用相同的逻辑。我的以下尝试在编译时失败,并出现以下错误:
“订阅”类型缺少“Observable
有了这个,我怎样才能同时返回一个可观察对象并在 http 拦截器上维护一个队列?我的方法有缺陷吗?我可以使用http拦截器来限制http速率吗?
@Injectable()
export class I1 implements HttpInterceptor {
intercept(
req: HttpRequest<any>,next: HttpHandler
): Observable<HttpEvent<any>> {
const modified = req.clone({ setHeaders: { "Custom-Header-1": "1" } });
return next
.handle(req)
.do((ev: HttpEvent<any>) => {
if (ev instanceof HttpResponse) {
console.log(ev);
}
})
.pipe(
bufferTime(1000,null,1),filter(buffer => buffer.length > 0),concatMap(buffer => of(buffer).pipe(delay(1000)))
)
.subscribe(console.log);
}
}
https://stackblitz.com/edit/angular-interceptors-npqkjp?file=app/interceptors.ts
解决方法
如果您想了解更多关于拦截器和 HttpClientModule 的工作原理,您可以查看这篇文章:Exploring the HttpClientModule in Angular。
我的方法有缺陷吗?
在这种情况下,问题在于 next.handle
应该返回一个 Observable,但是通过订阅它,它返回一个订阅。
为了更好地理解原因,我将粘贴从上面链接的文章中复制的片段:
const obsBE$ = new Observable(obs => {
timer(1000)
.subscribe(() => {
// console.log('%c [OBSERVABLE]','color: red;');
obs.next({ response: { data: ['foo','bar'] } });
// Stop receiving values!
obs.complete();
})
return () => {
console.warn("I've had enough values!");
}
});
// Composing interceptors the chain
const obsI1$ = obsBE$
.pipe(
tap(() => console.log('%c [i1]','color: blue;')),map(r => ({ ...r,i1: 'intercepted by i1!' }))
);
let retryCnt = 0;
const obsI2$ = obsI1$
.pipe(
tap(() => console.log('%c [i2]','color: green;')),map(r => {
if (++retryCnt <=3) {
throw new Error('err!')
}
return r;
}),catchError((err,caught) => {
return getRefreshToken()
.pipe(
switchMap(() => /* obsI2$ */caught),)
})
);
const obsI3$ = obsI2$
.pipe(
tap(() => console.log('%c [i3]','color: orange;')),i3: 'intercepted by i3!' }))
);
function getRefreshToken () {
return timer(1500)
.pipe(q
map(() => ({ token: 'TOKEN HERE' })),);
}
function get () {
return obsI3$
}
get()
.subscribe(console.log)
/*
-->
[i1]
[i2]
I've had enough values!
[i1]
[i2]
I've had enough values!
[i1]
[i2]
I've had enough values!
[i1]
[i2]
[i3]
{
"response": {
"data": [
"foo","bar"
]
},"i1": "intercepted by i1!","i3": "intercepted by i3!"
}
I've had enough values!
*/
要点是拦截器创建某种链,以负责发出实际请求的可观察对象结束。 This 是链中的最后一个节点:
return new Observable((observer: Observer<HttpEvent<any>>) => {
// Start by setting up the XHR object with request method,URL,and withCredentials flag.
const xhr = this.xhrFactory.build();
xhr.open(req.method,req.urlWithParams);
if (!!req.withCredentials) {
xhr.withCredentials = true;
}
/* ... */
})
如何同时在http拦截器返回一个observable并维护一个队列
我认为解决这个问题的一种方法是创建一个拦截器,该拦截器将包含队列逻辑并使其 intercept
方法返回一个 Observable
,以便它可以被订阅:
const queueSubject = new Subject<Observable>();
const pendingQueue$ = queueSubject.pipe(
// using `mergeAll` because the Subject's `values` are Observables
mergeAll(limit),share(),);
intercept (req,next) {
// `next.handle(req)` - it's fine to do this,no request will fire until the observable is subscribed
queueSubject.next(
next.handle(req)
.pipe(
// not interested in `Sent` events
filter(ev => ev instanceof HttpResponse),filter(resp => resp.url === req.url),)
);
return pendingQueue$;
}
使用 filter
运算符是因为通过使用 share
,响应将发送给所有订阅者。假设您同步调用 http.get
5 次,因此 share
的主题有 5 个新订阅者,最后一个将收到其响应,但也会收到其他请求的响应。因此,用户可以使用 filter
来为请求提供正确的响应,在这种情况下,通过将请求的 URL(req.url
) 与我们从 HttpResponse.url
获得的 URL 进行比较:
observer.next(new HttpResponse({
body,headers,status,statusText,url: url || undefined,}));
现在,我们为什么要使用 share()
?
让我们先看一个更简单的例子:
const s = new Subject();
const queue$ = s.pipe(
mergeAll()
)
function intercept (req) {
s.next(of(req));
return queue$
}
// making request 1
intercept({ url: 'req 1' }).subscribe();
// making request 2
intercept({ url: 'req 2' }).subscribe();
// making request 3
intercept({ url: 'req 3' }).subscribe();
此时,Subject s
应该有 3 个订阅者。这是因为当您返回队列时,您将返回 s.pipe(...)
,而当您订阅时,这与执行相同:
s.pipe(/* ... */).subscribe()
所以,这就是主题最后会有 3 个订阅者的原因。
现在让我们检查相同的代码段,但使用 share()
:
const queue$ = s.pipe(
mergeAll(),share()
);
// making request 1
intercept({ url: 'req 1' }).subscribe();
// making request 2
intercept({ url: 'req 2' }).subscribe();
// making request 3
intercept({ url: 'req 3' }).subscribe();
在您订阅请求 1 后,share
将创建一个 Subject 实例,所有后续订阅者都将属于它,而不是属于 main Subject s
。因此,s
将只有一个订阅者。这将确保我们正确实现队列,因为尽管主题 s
只有一个订阅者,但它仍然会接受 s.next()
值,其结果将传递给另一个主题(那个来自 share()
),它最终会将响应发送给其所有订阅者。
在拦截器上,您返回的是订阅,而不是 Observable。
如果删除 .subscribe(console.log)
行,它应该可以正常编译。订阅由消费者完成。
如果您想对发出的所有内容进行 console.log,请使用 tap(next => ...)
运算符
编辑 - 嗯,它解决了编译错误,但我不确定它会像你想要的那样工作......我不完全理解拦截器是如何工作的。