我想找到一种方法来检测观察者是否已经完成使用我使用Rx.Observable.create创建的自定义observable,这样定制的observable可以结束它并正确地进行一些清理.

因此,我创建了一些测试代码,如下所示,为此目的找出观察者对象上可用的字段类型.

var Rx = require("rx")

var source = Rx.Observable.create(function (observer) {

  var i = 0;
  setInterval(function(){
    observer.onNext(i);
    console.dir(observer);
    i+=1
  }, 1000)

});

var subscription = source.take(2).subscribe(
  function (x) { console.log('onNext: %s', x); }
);

输出如下

onNext: 0
{ isStopped: false,
  observer: 
   { isStopped: false,
     _onNext: [Function],
     _onError: [Function],
     _onCompleted: [Function] },
  m: { isDisposed: false, current: { dispose: [Function] } } }
onNext: 1
onCompleted
{ isStopped: true,
  observer: 
   { isStopped: false,
     _onNext: [Function],
     _onError: [Function],
     _onCompleted: [Function] },
  m: { isDisposed: true, current: null } }

似乎观察者对象上有3个字段似乎与我的目标有关,即observer.isStopped,observer.observer.isStopped和observer.m.isDiposed.

我想知道他们都是关于什么以及我应该选择哪一个.

================================================== ============================
我的问题的动机

根据Andre的建议,我添加了激发我的问题的场景.

在我的应用程序中,我试图基于window.requestAnimationFrame(回调)机制做一些UI动画. requestAnimationFrame将在浏览器渲染引擎确定的时间内调用提供的回调.回调应该做一些动画步骤,并再次递归调用requestAnimationFrame直到动画结束.

我想将此机制抽象为可观察的,如下所示.

function animationFrameRenderingEventsObservable(){
    return Rx.Observable.create(function(subscriber){
        var fn = function(frameTimestmpInMs){
            subscriber.onNext(frameTimestmpInMs);
            window.requestAnimationFrame(fn)
        };
        window.requestAnimationFrameb(fn);
    });
}

然后我可以在需要动画的各个地方使用它.举个例子,我需要绘制一些动画,直到用户触摸屏幕,我走了

 animationFrameRenderingEventsObservable()
   .takeUntil(touchStartEventObservable)
   .subscribe( animationFunc )

但是,在takeUntil(touchStartEventObservable)结束订阅之后,我需要一种方法来停止animationFrameRenderingEventsObservable中的无限递归.

因此,我将animationFrameRenderingEventsObservable修改为

function animationFrameRenderingEventsObservable(){
    return Rx.Observable.create(function(subscriber){
        var fn = function(frameTimestmpInMs){
            if (!subscriber.isStopped){
                subscriber.onNext(frameTimestmpInMs);
                window.requestAnimationFrame(fn)
            }else{
                subscriber.onCompleted();
            }
        };
        window.requestAnimationFrameb(fn);
    });
}

根据我的测试,代码按预期工作.但是,如果像安德烈提到的那样,使用subscriber.isStopped或类似的方法并不是正确的方法,那么正确的方法是什么？

解决方法:

在您提供给create的函数中,您可以返回一个清理函数,以便在观察者从您的observable取消订阅时调用.您应该提供一个停止动画帧请求的函数.这是一个可行的观察,可以完成我几年前写的你想要的东西：

Rx.Observable.animationFrames = function () {
    /// <summary>
    /// Returns an observable that triggers on every animation frame (see https://developer.mozilla.org/en-US/docs/Web/API/window.requestAnimationFrame ).
    /// The value that comes through the observable is the time(ms) since the previous frame (or the time since the subscribe call for the first frame)
    /// </summary>
    var request = window.requestAnimationFrame || window.mozRequestAnimationFrame || window.webkitRequestAnimationFrame || window.msRequestAnimationFrame,
        cancel = window.cancelAnimationFrame || window.mozCancelAnimationFrame || window.webkitCancelAnimationFrame || window.webkitCancelRequestAnimationFrame ||
            window.msCancelAnimationFrame || window.msCancelRequestAnimationFrame;

    return Rx.Observable.create(function (observer) {
        var requestId,
            startTime = window.mozAnimationStartTime || Date.now(),
            callback = function (currentTime) {
                // If we have not been disposed, then request the next frame
                if (requestId !== undefined) {
                    requestId = request(callback);
                }

                observer.onNext(Math.max(0, currentTime - startTime));
                startTime = currentTime;
            };

        requestId = request(callback);

        return function () {
            if (requestId !== undefined) {
                var r = requestId;
                requestId = undefined;
                cancel(r);
            }
        };
    });
};

用法：

Rx.Observable.animationFrames().take(5).subscribe(function (msSinceLastFrame) { ... });