"Unsubscribe" 函数 callback/hook 在 Observable "executor" 函数中

Question

我对 "dispose" 或 "unsubscribe" 函数的用途感到困惑，它是（可选地）从一个可观察的 "executor" 函数中 return 编辑的，像这样：

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();

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

任何人都可以告诉我在那种情况下 returning 一个函数的目的是什么，这个函数被调用的是什么，它的签名是什么？我无法弄清楚有关它的信息。

我还看到了 return 从执行程序函数订阅的更复杂的示例，例如：

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

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

把问题说清楚，这样做有什么区别：

const sub = new Rx.Subject();

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

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

});

而不是 return 订阅调用的结果：

const sub = new Rx.Subject();

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

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

});

Answer 1

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

关于你的问题； .subscribe() return 是您可以调用 .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>

"Unsubscribe" 函数 callback/hook 在 Observable "executor" 函数中

"Unsubscribe" function callback/hook in Observable "executor" function

javascript

rxjs5