我有与Add queueing to angulars $http service 中提到的完全相同的要求，但需要使用来自@angular/common/http 的HttpInterceptor 在Angular 4.3 或5 中实现。
我有一个非常古怪的 API，它一次只能处理一个特定浏览器会话的请求。因此，我需要确保每次在同一个会话中发出请求时，它都会进入一个队列，并且该队列一次执行一个请求，直到它为空。
【问题讨论】：
标签： angular observable
@Zlatko 提出了正确的方法，虽然其中几乎没有逻辑和语法问题，但我已将其粘贴在工作代码下方：
import { Injectable } from '@angular/core';
import { Response } from '@angular/http';
import { HttpClient } from '@angular/common/http';
import { Observable } from 'rxjs/Observable';
import { Subject } from 'rxjs/Subject'
export class PendingRequest {
url: string;
method: string;
options: any;
subscription: Subject<any>;
constructor(url: string, method: string, options: any, subscription: Subject<any>) {
this.url = url;
this.method = method;
this.options = options;
this.subscription = subscription;
}
}
@Injectable()
export class BackendService {
private requests$ = new Subject<any>();
private queue: PendingRequest[] = [];
constructor(private httpClient: HttpClient) {
this.requests$.subscribe(request => this.execute(request));
}
/** Call this method to add your http request to queue */
invoke(url, method, params, options) {
return this.addRequestToQueue(url, method, params, options);
}
private execute(requestData) {
//One can enhance below method to fire post/put as well. (somehow .finally is not working for me)
const req = this.httpClient.get(requestData.url)
.subscribe(res => {
const sub = requestData.subscription;
sub.next(res);
this.queue.shift();
this.startNextRequest();
});
}
private addRequestToQueue(url, method, params, options) {
const sub = new Subject<any>();
const request = new PendingRequest(url, method, options, sub);
this.queue.push(request);
if (this.queue.length === 1) {
this.startNextRequest();
}
return sub;
}
private startNextRequest() {
// get next request, if any.
if (this.queue.length > 0) {
this.execute(this.queue[0]);
}
}
}
可以使用/调用上述服务以下列方式进行 HTTP 调用（来自任何其他组件或服务）（显然您需要在组件中注入 BackendService，例如在组件的提供者中提及并在构造函数中定义）：
this.backendService.invoke('https://jsonplaceholder.typicode.com/posts', 'Get', null, null)
.subscribe(
result => {
this.data = result;
}
);
如果有人想看看工作的 plunker，那么here is the working plunker。
【讨论】：
您可以相对轻松地做到这一点。下面是一个简单的例子。
它缺少打字等，没有详细说明，它有一些弱点，最好将排队部分和 http-requesting 部分提取到不同的服务或类中，但这应该可以帮助您入门。
interface PendingRequest {
url: string;
method: string;
options: any;
subscription: Observable<any>;
}
@Injectable()
export class BackendService {
// This is your private requests queue emitter.
private requests$: Subject = new Subject();
private queue: PendingRequest[] = [];
constructor(private http: HttpClient) {
// subscribe to that queue up there
this.requests$.subscribe(request => this.execute(request));
}
// This is your public API - you can extend it to get/post/put or specific
// endpoints like 'getUserProfile()' etc.
invoke(url, method, options) {
return this.addRequestToQueue(url, method, params, options);
}
private execute(requestData) {
const req = this.httpClient.request(requestData.method, requestData.url, requestData.options)
// as a last step, invoke next request if any
.finally(() => this.startNextRequest());
const sub = requestData.subscription;
sub.switchMap(req);
}
private addRequestToQueue(url, method, options) {
const sub = new Subject<any>();
const request = new PendingRequest(url, method, options, sub)
// if there are no pending req's, execute immediately.
if (this.queue.length === 0) {
this.requests$.next(request);
} else {
// otherwise put it to queue.
this.queue.push(request);
}
return sub;
}
private startNextRequest() {
// get next request, if any.
if (this.queue.length) {
this.execute(this.queue.shift());
}
}
}
【讨论】：
我的要求和你一模一样。其他答案工作得很好，只是它需要开发人员使用另一个自定义服务而不是本机 HttpClient 创建请求。您也可以尝试以下拦截器来应用排队。
此解决方案需要您添加 2 个服务，一个 HttpInterceptor 和一个服务 (RequestQueueService) 来管理队列。
HttpInterceptor：
@Injectable()
export class QueueInterceptorService implements HttpInterceptor {
constructor(private queueService: RequestQueueService) { }
intercept(request: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
return this.queueService.intercept(request, next);
}
}
请求队列服务：
@Injectable({
providedIn: 'root'
})
export class RequestQueueService {
private queue: ReplaySubject<any>[] = [];
intercept(request: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
const requestQueueItem$ = new ReplaySubject<any>();
const result$ = requestQueueItem$.pipe(
switchMap(() => next.handle(request).pipe(
tap(req => {
if (req.type == HttpEventType.Response) {
this.processNextRequest();
}
}),
catchError(err => {
this.processNextRequest();
throw err;
})
))
);
this.queue.push(requestQueueItem$);
if (this.queue.length <= 1) {
this.dispatchRequest();
}
return result$;
}
private processNextRequest(): void {
if (this.queue && this.queue.length > 0) {
this.queue.shift();
}
this.dispatchRequest();
}
private dispatchRequest(): void {
if (this.queue.length > 0) {
const nextSub$ = this.queue[0];
nextSub$.next();
nextSub$.complete();
}
}
}
最后，在 AppModule 中：
@NgModule({
declarations: [
AppComponent
],
imports: [
BrowserModule,
HttpClientModule
],
providers: [
RequestQueueService,
{ provide: HTTP_INTERCEPTORS, useClass: QueueInterceptorService, multi: true }
],
bootstrap: [AppComponent]
})
export class AppModule { }
顺便说一句，我正在使用 Angular 8 和 rxjs 6.4。
【讨论】：
Ryan Teh 的 Answer 如果您想在一个提示中概括所有 http 请求，它就像一个魅力，因为它是一个 Catch All 解决方案。
在这里，我使用 Angular 8.2.4 和 rxjs 6.4.0 的足够导入来扩展它，因为会有很多用户在寻找 cut n' paste
对于创建新答案而不是评论，我深表歉意，我仍然没有足够的信任。
queue-interceptor.service.ts
import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';
import { HttpInterceptor, HttpRequest, HttpHandler, HttpEvent } from '@angular/common/http';
import { RequestQueueService } from './request-queue.service';
@Injectable()
export class QueueInterceptorService implements HttpInterceptor {
constructor(private queueService: RequestQueueService) { }
intercept(request: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
return this.queueService.intercept(request, next);
}
}
request-queue.service.ts
import { Injectable } from '@angular/core';
import { HttpRequest, HttpHandler, HttpEvent, HttpEventType } from '@angular/common/http';
import { Observable, ReplaySubject } from 'rxjs';
import { tap, catchError, switchMap } from 'rxjs/operators';
@Injectable({
providedIn: 'root'
})
export class RequestQueueService {
private queue: ReplaySubject<any>[] = [];
intercept(request: HttpRequest<any>, next: HttpHandler): Observable<HttpEvent<any>> {
const requestQueueItem$ = new ReplaySubject<any>();
const result$ = requestQueueItem$.pipe(
switchMap(() => next.handle(request).pipe(
tap(req => {
if (req.type == HttpEventType.Response) {
this.processNextRequest();
}
}),
catchError(err => {
this.processNextRequest();
throw err;
})
))
);
this.queue.push(requestQueueItem$);
if (this.queue.length <= 1) {
this.dispatchRequest();
}
return result$;
}
private processNextRequest(): void {
if (this.queue && this.queue.length > 0) {
this.queue.shift();
}
this.dispatchRequest();
}
private dispatchRequest(): void {
if (this.queue.length > 0) {
const nextSub$ = this.queue[0];
nextSub$.next();
nextSub$.complete();
}
}
}
和 app.module.ts
import { NgModule } from '@angular/core';
import { BrowserModule } from '@angular/platform-browser';
import { AppComponent } from './app.component';
import { HttpClientModule, HTTP_INTERCEPTORS } from '@angular/common/http';
import { RequestQueueService } from './YOUR_SERVICES_DIRECTORY/request-queue.service';
import { QueueInterceptorService } from './YOUR_SERVICES_DIRECTORY/queue-interceptor.service';
@NgModule({
declarations: [
AppComponent
],
imports: [
BrowserModule,
HttpClientModule
],
providers: [
RequestQueueService,
{ provide: HTTP_INTERCEPTORS, useClass: QueueInterceptorService, multi: true }
],
bootstrap: [AppComponent]
})
export class AppModule { }
【讨论】：