如何解决Angular - http 拦截器 - http 速率限制器 - 滑动窗口
我有一个用例，我需要限制传出 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 => ...) 运算符
编辑 - 嗯，它解决了编译错误，但我不确定它会像你想要的那样工作......我不完全理解拦截器是如何工作的。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
