将无限异步回调序列转换为可观察序列？

Question

假设我有以下基于异步回调的“无限”序列，我会在一段时间后取消它：

'use strict';

const timers = require('timers');

let cancelled = false;

function asyncOperation(callback) {
  const delayMsec = Math.floor(Math.random() * 10000) + 1;
  console.log(`Taking ${delayMsec}msec to process...`);
  timers.setTimeout(callback, delayMsec, null, delayMsec);
}

function cancellableSequence(callback) {
  asyncOperation((error, processTime) => {
    console.log('Did stuff');
    if (!cancelled) {
      process.nextTick(() => { cancellableSequence(callback); });
    } else {
      callback(null, processTime);
    }
  });
}

cancellableSequence((error, lastProcessTime) => {
  console.log('Cancelled');
});

timers.setTimeout(() => { cancelled = true; }, 0);

asyncOperation 将执行并至少回调一次，取消消息不会立即显示，而是在asyncOperation 完成后显示。对asyncOperation 的调用次数取决于内部delayMsec 值和最后传递给setTimeout() 的延迟参数（试图表明它们是可变的）。

我开始学习 RxJS5，并认为可以将其转换为 Observable 序列（“哦，Observable 订阅可以是 unsubscribe()d - 看起来很整洁！”）。

但是，我尝试将 cancellableSequence 转换为 ES6 生成器（否则如何实现无限？）产生 Observable.bindNodeCallback(asyncOperation)() 导致立即产生，在我的情况下这是不希望的行为。

我不能使用Observable.delay() 或Observable.timer()，因为我没有已知的一致间隔。 （asyncOperation 中的 Math.random(...) 试图表明我作为调用者不控制时间，并且回调发生在“某个未知时间之后”。）

我的失败尝试：

'use strict';

const timers = require('timers');
const Rx = require('rxjs/Rx');

function asyncOperation(callback) {
  const delayMsec = Math.floor(Math.random() * 10000) + 1;
  console.log(`Taking ${delayMsec}msec to process...`);
  timers.setTimeout(callback, delayMsec, null, delayMsec);
}

const operationAsObservable = Rx.Observable.bindNodeCallback(asyncOperation);
function* generator() {
  while (true) {
    console.log('Yielding...');
    yield operationAsObservable();
  }
}

Rx.Observable.from(generator()).take(2).mergeMap(x => x).subscribe(
  x => console.log(`Process took: ${x}msec`),
  e => console.log(`Error: ${e}`),
  c => console.log('Complete')
)

哪个结果是输出：

Yielding...
Taking 2698msec to process...
Yielding...
Taking 2240msec to process...
Process took: 2240msec
Process took: 2698msec
Complete

收益立即发生。 Process took: xxx 输出出现在您预期的时间（分别在 2240 和 2698 毫秒之后）。

（平心而论，我关心收益率之间的延迟的原因是 asyncOperation() 实际上是一个限速令牌桶库，它控制异步回调的速率 - 我想实现保留。）

顺便说一句，我试图用延迟取消替换take(2)，但从未发生过：

const subscription = Rx.Observable.from(generator()).mergeMap(x => x).subscribe(
  x => console.log(`Process took: ${x}msec`),
  e => console.log(`Error: ${e}`),
  c => console.log('Complete')
)

console.log('Never gets here?');
timers.setTimeout(() => {
  console.log('Cancelling...');
  subscription.unsubscribe();
}, 0);

可以通过 RxJS 取消订阅来完成我的尝试吗？ （我可以看到其他方法，例如process.exec('node', ...) 将asyncOperation() 作为单独的进程运行，让我能够process.kill(..) 等，但我们不要去那里......）。

我最初的基于回调的实现是实现可取消序列的建议方法吗？

更新的解决方案：

请参阅下面我对@user3743222 的回答的回复评论。这是我最终得到的结果（将 ES6 生成器替换为 Observable.expand()）：

'use strict';

const timers = require('timers');
const Rx = require('rxjs/Rx');

function asyncOperation(callback) {
  const delayMsec = Math.floor(Math.random() * 10000) + 1;
  console.log(`Taking ${delayMsec}msec to process...`);
  timers.setTimeout(callback, delayMsec, null, delayMsec);
}

const operationAsObservable = Rx.Observable.bindNodeCallback(asyncOperation);

const subscription = Rx.Observable
  .defer(operationAsObservable)
  .expand(x => operationAsObservable())
  .subscribe(
    x => console.log(`Process took: ${x}msec`),
    e => console.log(`Error: ${e}`),
    c => console.log('Complete')
  );

subscription.add(() => {
  console.log('Cancelled');
});

timers.setTimeout(() => {
  console.log('Cancelling...');
  subscription.unsubscribe();
}, 0);

更新的解决方案 2：

这是我为备用 RxJS4 repeatWhen() 方法提出的建议：

'use strict';

const timers = require('timers');
const Rx = require('rx');

function asyncOperation(callback) {
  const delayMsec = Math.floor(Math.random() * 1000) + 1;
  console.log(`Taking ${delayMsec}msec to process...`);
  timers.setTimeout(callback, delayMsec, null, delayMsec);
}

const operationAsObservable = Rx.Observable.fromNodeCallback(asyncOperation);

const subscription = Rx.Observable
  .defer(operationAsObservable)
  .repeatWhen(x => x.takeWhile(y => true))
  .subscribe(
    x => console.log(`Process took: ${x}msec`),
    e => console.log(`Error: ${e}`),
    c => console.log('Complete')
  );

timers.setTimeout(() => {
  console.log('Cancelling...');
  subscription.dispose();
}, 10000);

Answer 1

每次完成时，您似乎都在重复一个动作。对于expand 或repeatWhen，这看起来是一个很好的用例。

通常情况下，会是这样的：

Rx.Observable.just(false).expand(_ => {  
  return cancelled ? Rx.Observable.empty() : Rx.Observable.fromCallback(asyncAction)
})

您在任何时间点将cancelled 设置为true，当当前操作完成时，它会停止循环。还没有测试过，所以我很想知道这最终是否有效。

您可以查看有关投票的类似问题：

• How to build an rx poller that waits some interval AFTER the previous ajax promise resolves?

文档：

• [fromCallback]
• [expand]

文档链接适用于 Rxjs 4，但与 v5 相比应该没有太大变化