RXJS - 带有 .startWith 的 IntervalObservable答案

【问题标题】：RXJS - IntervalObservable with .startWithRXJS - 带有 .startWith 的 IntervalObservable
【发布时间】：2017-10-03 22:33:13
【问题描述】：

我已经在 Angular 应用程序中使用 IntervalObservable 和 startWith 实现了 http 池以立即启动。我想知道 IntervalObservable 是否等到初始/上一个调用完成流数据？有没有更好的方法在 Angular 应用中实现数据池。

来自 service.ts

getRecordsList() {
  return IntervalObservable
    .create(15000)
    .startWith(0)
    .flatMap((r) => this.http
    .post(`http://services.com/restful/recordService/getRecordsList`, body, {
      headers: new HttpHeaders().set('Content-Type', 'application/json')
    }))
    .shareReplay()
    .catch(this.handleError);
}

来自 component.ts 的示例

ngOnInit() {
 this.service.getRecordsList()
  .subscribe(
    (recordList) =>  {
      this.recordResponse = recordList;          
    },
    error => { console.log },
    () => console.log("HTTP Observable getRecordsList() completed...")
);

}

我使用过 Angular httClient，我希望无论如何这都无关紧要。

【问题讨论】：

标签： angular rxjs angular-httpclient

【解决方案1】：

以下内容可能对您有所帮助：

const { Observable } = Rx;

// HTTP MOCK
const http = {
  post: () => Observable.of('some response').delay(1000)
}

// SERVICE PART
const polling$ = Observable.timer(0, 5000);

const myRequest$ = polling$
  .do(() => console.log(`launching a new request...`))
  .switchMap(() => http.post('some-url'));

// COMPONENT PART
myRequest$
  .do(res => console.log(`Response: ${res}`))
  .subscribe();

还有一个正常工作的 Plunkr：https://plnkr.co/edit/BCTmlOv6FarNN1iUmMcA?p=preview

【讨论】：

【解决方案2】：

您的代码对我来说似乎是合理的。这是一个类似的代码，它使用IntervalObservable 将服务器池化，直到满足某些条件。

import { Component } from '@angular/core';
import { Http } from '@angular/http';
import { IntervalObservable } from 'rxjs/observable/IntervalObservable';
import 'rxjs/add/operator/takeWhile';
import 'rxjs/add/operator/startWith';

@Component({
    selector: 'my-angular2-app',
    templateUrl: './tool-component.html',
    styleUrls: ['./tool-component.css']
})
export class ToolComponent {

    private _APIBaseURL = 'https://api.app.io';
    private _autoRefresh: boolean = true;
    private _downloadLink: string = ""
    private _fileID: string = ""

    constructor(private _http: Http) { }
    // add code here
    // ...

    getDownloadLink() {
        this._autoRefresh = true;
        this._downloadLink = "";
        IntervalObservable
            .create(10000)
            .startWith(0)
            .takeWhile(() => this._autoRefresh)
            .subscribe(() => {
                this._http.get(`${this._APIBaseURL}/rocess?file-name=${this._fileID}`)
                    .subscribe(data => {
                        let idata = data.json();
                        if (idata['current_status'] == "done") {
                            this._downloadLink = idata.url;
                            this._autoRefresh = false;
                        }
                    })
            }
            )
    }
}

【讨论】：