javascript - "Unsubscribe"函数回调/Observable 中的钩子(Hook) "executor"函数

Question

我对“dispose”或“unsubscribe”函数的目的感到困惑，它(可选)从可观察的“executor”函数返回，如下所示:

const Rx = require('rxjs');

const obs = Rx.Observable.create(obs => {

    // we are in the Observable "executor" function
    obs.next(4);

     // we return this function, which gets called if we unsubscribe
    return function () {
        console.log('disposed');
    }

});

    const s1 = obs.subscribe(
        function (v) {
            console.log(v);
        },
        function (e) {
            console.log(e);
        },
        function () {
            console.log('complete');
        }
    );

    const s2 = obs.subscribe(
        function (v) {
            console.log(v);
        },
        function (e) {
            console.log(e);
        },
        function () {
            console.log('complete');
        }
    );


    s1.unsubscribe();
    s2.unsubscribe();

让我感到困惑的是，这样的函数实际上更有可能保留代码中的引用，从而防止垃圾回收。

谁能告诉我在那种情况下返回函数的目的是什么，函数被调用的是什么，它的签名是什么？我无法找出有关它的信息。

我还看到了从执行函数返回订阅的更复杂的示例，例如:

    let index = 0;

    let obsEnqueue = this.obsEnqueue = new Rx.Subject();

    this.queueStream = Rx.Observable.create(obs => {

        const push = Rx.Subscriber.create(v => {
            if ((index % obsEnqueue.observers.length) === obsEnqueue.observers.indexOf(push)) {
                obs.next(v);
            }
        });

        return obsEnqueue.subscribe(push);
    });

这似乎返回了一个订阅，而不仅仅是一个简单的函数。谁能解释一下这是怎么回事？

为了明确问题，这样做有什么区别:

const sub = new Rx.Subject();

const obs = Rx.Observable.create($obs => {

    $obs.next(4);
    return sub.subscribe($obs);

});

并且不返回订阅调用的结果:

const sub = new Rx.Subject();

const obs = Rx.Observable.create($obs => {

    $obs.next(4);
    sub.subscribe($obs);

});

Answer 1

Rx.Observable.create 需要返回的unsubscribe 函数在下游不再监听流时被调用，有效地让你有时间清理资源。

关于您的问题； .subscribe() 返回您可以调用 .unsubscribe() 的订阅。因此，如果您想对其他订阅执行某些操作，您可以将该订阅通过管道传输到您的下游:

    const obs = Rx.Observable.create($obs => {
      const timer = Rx.Observable.interval(300)
        .do(i => console.log('emission: ' + i))

      return timer.subscribe($obs);
    });
    obs.take(4).subscribe(i => console.log('outer-emission:'+i))

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.2/Rx.js"></script>

如果没有取消订阅功能，您将停止收听可观察对象，但内部创建的间隔将继续运行:

const obs = Rx.Observable.create($obs => {
  const timer = Rx.Observable.interval(300)
    .do(i => console.log('emission: ' + i))
    .take(10)
    .subscribe(
      val => $obs.next(val),
      err => $obs.error(err),
      () => $obs.complete()
    );

  return function(){} // empty unsubscribe function, internal subscription will keep on running
});
obs.take(4).subscribe(i => console.log('outer-emission:'+i))

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.2/Rx.js"></script>