使用Angular2/RxJS读取缓冲响应

Question

我正在建立一个从后端读取数据的网站.该数据即时计算并以缓冲方式发送回客户端.即,一旦计算出第一个块,它就会被发送到客户端,然后它会计算下一个块并将其发送给客户端.整个过程发生在同一个HTTP请求中.客户端不应该等待完成响应完成,而是在发送后立即自己处理每个块.通常可以使用XHR进程处理程序(例如,如何从XMLHttpRequest获取进度)来使用此类响应.

如何使用RxJS和Observables在Angular2中使用HttpModule消耗这样的响应？

---

编辑:peeskillet在下面给出了一个非常详细的答案.此外,我做了一些进一步的挖掘,发现了一个关于Angular和StackOverflow问题的功能请求HttpModule,另一个方法是如何解决它.

Answer 1

注意:以下答案仅为POC.它旨在教育Http的体系结构,并提供简单的工作POC实现.我们应该看看XHRConnection有关在实施时应该考虑的其他方面的想法的来源.

在尝试实现这一点时,我认为没有任何方法可以直接进入XHR.似乎我们需要提供一些与使用相关的组件的自定义实现Http.我们应该考虑的三个主要组成部分是

• Connection
• ConnectionBackend
• Http

Http将一个ConnectionBackend参数作为其构造函数的参数.当发出请求时,比如说get,Http创建一个连接ConnectionBackend.createConnection,并返回(从中返回)的Observable属性.在最简单的(简化)视图中,它看起来像这样ConnectioncreateConnection

class XHRConnection implements Connection {
  response: Observable<Response>;
  constructor( request, browserXhr) {
    this.response = new Observable((observer: Observer<Response>) => {
      let xhr = browserXhr.create();
      let onLoad = (..) => {
        observer.next(new Response(...));
      };
      xhr.addEventListener('load', onLoad);
    })
  }
}

class XHRBackend implements ConnectionBackend {
  constructor(private browserXhr) {}
  createConnection(request): XHRConnection {
    return new XHRConnection(request, this.broswerXhr).response;
  }
}

class Http {
  constructor(private backend: ConnectionBackend) {}

  get(url, options): Observable<Response> {
    return this.backend.createConnection(createRequest(url, options)).response;
  }
}

因此,了解这种架构,我们可以尝试实现类似的东西.

对于Connection,这是POC.为简洁而省略了进口,但在大多数情况下,所有东西都可以从中导入@angular/http,并且Observable/Observer可以从中导入rxjs/{Type}.

export class Chunk {
  data: string;
}

export class ChunkedXHRConnection implements Connection {
  request: Request;
  response: Observable<Response>;
  readyState: ReadyState;

  chunks: Observable<Chunk>;

  constructor(req: Request, browserXHR: BrowserXhr, baseResponseOptions?: ResponseOptions) {
    this.request = req;
    this.chunks = new Observable<Chunk>((chunkObserver: Observer<Chunk>) => {
      let _xhr: XMLHttpRequest = browserXHR.build();
      let previousLen = 0;
      let onProgress = (progress: ProgressEvent) => {
        let text = _xhr.responseText;
        text = text.substring(previousLen);
        chunkObserver.next({ data: text });
        previousLen += text.length;

        console.log(`chunk data: ${text}`);
      };
      _xhr.addEventListener('progress', onProgress);
      _xhr.open(RequestMethod[req.method].toUpperCase(), req.url);
      _xhr.send(this.request.getBody());
      return () => {
        _xhr.removeEventListener('progress', onProgress);
        _xhr.abort();
      };
    });
  }
}

这是我们刚订阅XHR progress活动.由于XHR.responseText喷出整个连接文本,我们只是substring得到块,并通过发出每个chuck Observer.

对于XHRBackend,我们有以下(没什么了不起的).同样,一切都可以从@angular/http;

@Injectable()
export class ChunkedXHRBackend implements ConnectionBackend {
  constructor(
      private _browserXHR: BrowserXhr, private _baseResponseOptions: ResponseOptions,
      private _xsrfStrategy: XSRFStrategy) {}

  createConnection(request: Request): ChunkedXHRConnection {
    this._xsrfStrategy.configureRequest(request);
    return new ChunkedXHRConnection(request, this._browserXHR, this._baseResponseOptions);
  }
}

对于Http,我们将扩展它,添加一个getChunks方法.如果需要,您可以添加更多方法.

@Injectable()
export class ChunkedHttp extends Http {
  constructor(protected backend: ChunkedXHRBackend, protected defaultOptions: RequestOptions) {
    super(backend, defaultOptions);
  }

  getChunks(url, options?: RequestOptionsArgs): Observable<Chunk> {
    return this.backend.createConnection(
       new Request(mergeOptions(this.defaultOptions, options, RequestMethod.Get, url))).chunks;
  }
}

该mergeOptions方法可以在Http源中找到.

现在我们可以为它创建一个模块.用户应该直接使用ChunkedHttp而不是Http.但是因为不要尝试覆盖Http令牌,所以Http如果需要,仍然可以使用.

@NgModule({
  imports: [ HttpModule ],
  providers: [
    {
      provide: ChunkedHttp,
      useFactory: (backend: ChunkedXHRBackend, options: RequestOptions) => {
        return new ChunkedHttp(backend, options);
      },
      deps: [ ChunkedXHRBackend, RequestOptions ]
    },
    ChunkedXHRBackend
  ]
})
export class ChunkedHttpModule {
}

我们导入它HttpModule因为它提供了我们需要注入的其他服务,但是如果我们不需要,我们不希望重新实现它们.

测试只是导入ChunkedHttpModule到AppModule.另外要测试我使用了以下组件

@Component({
  selector: 'app',
  encapsulation: ViewEncapsulation.None,
  template: `
    <button (click)="onClick()">Click Me!</button>
    <h4 *ngFor="let chunk of chunks">{{ chunk }}</h4>
  `,
  styleUrls: ['./app.style.css']
})
export class App {
  chunks: string[] = [];

  constructor(private http: ChunkedHttp) {}

  onClick() {
    this.http.getChunks('http://localhost:8080/api/resource')
      .subscribe(chunk => this.chunks.push(chunk.data));
  }
}

我有一个后端端点设置,它"Message #x"每隔半秒就会以10个块的形式吐出.这就是结果

某处似乎有一个bug.只有九个 :-).我认为它与服务器端有关.