javascript - 如何取消订阅或处理 Angular2 或 RxJS 中的条件可观察间隔？

Question

我对整个 Rx 事物和响应式编程都是新手，但是我必须处理这种情况。我想要一个可观察的间隔，通过每 500 毫秒向其 REST API 发送一个 POST 请求来检查硬件的状态，以查看响应是否发生变化。因此，一旦它发生变化，我希望这种间隔可观察的 POST 请求立即关闭，将资源留给其他 future 的操作。这是一段代码。

myLoop(item_array:number[], otheroption : string){
    for (let item of item_array){
        //set hardware option, still a request
        this.configHardware(item,otheroption)
        //after config the hardware, command hardware take action
            .flatMap(() => {
                //this return a session id for check status
                this.takeHardwareAction()
                    .flatMap( (result_id) => {
                        //I want every 500ms check if hardware action is done or not
                        let actionCheckSubscription = IntervalObservable.create(500).subscribe(
                            () => {
                                //So my question is, can I unsubscribe the actionCheckSubscription in here on condition change? For example, 
                                if (this.intervalCheckingStatus(result_id))
                                    actionCheckSubscription.unsubscribe() ;
                            }
                        ) ;
                    })

            })
    }
}

Answer 1

所以你想每 500 毫秒发出一个 POST 请求，然后检查它的响应。我假设您的方法 intervalCheckingStatus 评估 POST 响应并确定它是否不同？

首先，我不会使用 IntervalObservable。你导入了 RxJS 模块了吗？它是 Angular 认可的第三方库，也是他们在所有开发人员指南示例中使用的库。如果没有，请安装并导入。 https://github.com/Reactive-Extensions/RxJS

import * as Rx from 'rxjs/Rx';

我假设您已经导入了 Http、ResponseOptions 等，但这里是为了以防其他人好奇:

import { Http, Response, ResponseOptions } from '@angular/http';

编辑 1: 忘记包含依赖项注入(inject)。将 Http 注入(inject)您的构造函数。我将其命名为 http，因此我将其命名为 this.http.post

constructor(private http: Http) {

然后，我将执行以下操作:

编辑 2: 这将在您的循环内，其中 post 参数与数组中的项目相关。

    // Every 500 ms, make a POST request
    Rx.Observable.interval(500)
                  // Add your POST arguments here
                 .map(_ => this.http.post(yourUrl, yourBody))
                 // This filter here is so that it will only emit when intervalCheckingStatus returns true
                 // You need to get the property you need from the Response: resp
                 // Is it the status code you're interested in? That's what I put as an example here but whatever it is, pass it to your method
                .filter(resp => this.intervalCheckingStatus(resp.status))
                // Take(1) takes only the first emitted value and once it does that, the observable completes. So you do NOT need to unsubscribe explicitly.
                .take(1);

如果您需要在响应具有您要查找的状态(或任何属性)后执行某些操作，则将 .subscribe 链接到末尾并在其中执行您需要的操作。同样，由于 take(1)，一旦第一个元素被抽取，可观察流就会完成，您不需要取消订阅。

此外，这里还有一个非常有用的网站: http://rxmarbles.com/#take您可以在他们的示例中看到，在采用 2 个元素后，生成的可观察对象是完整的(垂直线)。