Angular - http 拦截器 - http 速率限制器 - 滑动窗口

Question

我有一个用例，我需要限制传出 http 请求的数量。是的，我确实在服务器端有速率限制器，但前端也需要限制活动 http 请求的数量。因此，我正在尝试实现一个滑动窗口协议，在任何时候我都会只有 n 个活跃请求。

这种使用 Rxjs 的方法通常工作得很好，请参见此处： https: //jsbin.com/pacicubeci/1/edit ?js,console,output

但我不清楚如何对 http 拦截器使用相同的逻辑。我的以下尝试在编译时失败，并出现以下错误：

“Subscription”类型缺少“Observable<HttpEvent>”类型中的以下属性：_isScalar、source、operator、lift 以及其他 114 个属性。(2740)

这样，我怎样才能返回一个可观察对象并同时在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

Answer 1

如果您想了解更多关于拦截器和 HttpClientModule 如何在幕后工作的信息，您可以查看这篇文章：探索 Angular 中的 HttpClientModule。

我的方法有缺陷吗？在这种情况下，问题是next.handle期望返回一个 Observable，但是通过订阅它，它返回一个 Subscription。

为了更好地理解原因，我将粘贴从上面链接的文章中复制的片段：

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;')),
    map(r => ({ ...r, 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!
*/

StackBlitz 演示。

要点是拦截器创建某种链，该链以负责发出实际请求的可观察值结束。这是链中的最后一个节点：

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 拦截器中维护一个队列

我认为解决这个问题的一种方法是创建一个拦截器，其中包含队列逻辑并使其intercept方法返回 an 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.get5 次，因此有 5 个新的 Subscriber 订阅者share，最后一个订阅者将收到其响应，但也会收到其他请求的响应。因此，可以使用 can usefilter来为请求提供正确的响应，在本例中，通过将 request( req.url) 的 URL 与我们从 中获得的 URL进行比较HttpResponse.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();

此时，Subjects应该有 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实例，所有后续订阅者都将属于它，而不是属于主Subject s。因此，s将只有一名订户。这将确保我们正确实现队列，因为虽然主题s只有一个订阅者，但它仍然会接受s.next()值，其结果将传递给另一个主题（来自 的那个share()），后者最终将发送响应给所有订阅者。