RxJs

Question

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

因此，我创建了一些测试代码，如下所示，以确定观察者对象上有哪些类型的字段可用于此目的。

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。

我想知道它们都是关于什么的，我应该选择哪一个。

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

根据安德烈的建议，我添加了激发我的问题的场景。

在我的应用程序中，我试图根据 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);
    });
}

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

Answer 1

如果你使用 observer.isStopped 或相关的，你做错了什么。这些不是 API 函数，它们是实现细节。

I'd like to find a way to detect whether a observer has finished using a customized observable, which I created with Rx.Observable.create, such that the customized observable can end it and do some clean up properly.

观察者会在 'onCompleted' 发生时进行清理。在上面的 Observable.create 中，当您认为自定义可观察对象已经结束时，您应该调用 observer.onCompleted()，或者如果自定义可观察对象是无限的，则永远不要调用 observer.onCompleted()。此外，您应该将整个代码用 try/catch 包装在 Observable.create 中，并在 catch 中调用 observer.onError(err)。

如果您试图在观察者 "has finished using the observable" 时使 Observable "clean up"，那么您做错了。本质上，如果自定义可观察对象需要对观察者做出反应，那么这意味着观察者也应该是一个可观察对象。 Observable.create 可能不是这个工具。

最好说出你想要完成什么，而不是如何做一些具体的事情。

更新：

根据你想做的动画：在RxJS的上下文中，requestAnimationFrame是一个Scheduler，而不是一个Observable。 Use it from the RxJS-DOM library.

Answer 2

在您提供给 create 的函数中，您可以 return 当观察者取消订阅您的可观察对象时调用的清理函数。您应该提供一个函数来停止您的动画帧请求。这是我几年前写的一个可以工作的可观察对象：

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) { ... });

RxJs - observer.isStopped、observer.observer.isStopped 和 observed.m.isDisposed 之间有什么区别

RxJs - what's the difference among observer.isStopped, observer.observer.isStopped and observed.m.isDisposed

javascript