javascript - 重复/重置可观察值

Question

我正在使用 rxjs 为智能电视上的 Remote 创建“ channel 号”选择器。这个想法是，当您输入数字时，您会在屏幕上看到它们，并且在输入完数字后，用户实际上会被带到该 channel 。

我使用两个可观察量来实现此目的:

1. 一个“进度”流，用于监听所有数字输入，并在通过扫描运算符输入数字时发出串联的数字字符串。

2. “已完成”流，在 n 毫秒内未输入任何数字后，将发出已完成的最终数字字符串。例如:1-2-3 ->“123”。

这是我用来尝试解决此问题的代码:

channel 号:

module.exports = function (numberKeys, source, scheduler) {
    return function (completedDelay) {
        var toNumericString = function (name) {
                return numberKeys.indexOf(name).toString();
            },
            concat = function (current, numeric) {
                return current.length === 3 ? current : current + numeric;
            },
            live = createPress(source, scheduler)(numberKeys)
                .map(toNumericString)
                .scan(concat, '')
                .distinctUntilChanged(),

            completed = live.flatMapLatest(function (value) {
                return Rx.Observable.timer(completedDelay, scheduler).map(value);
            }),
            progress = live.takeUntil(completed).repeat();

        return {
            progress: progress,
            completed: completed
        };
    };
};

创建新闻:

module.exports = function (source, scheduler) {
    return function (keyName, throttle) {
        return source
            .filter(H.isKeyDownOf(keyName))
            .map(H.toKeyName);
    };
};

创建源:

module.exports = function (provider) {
    var createStream = function (event) {
        var filter = function (e) {
                return provider.hasCode(e.keyCode);
            },
            map = function (e) {
                return {
                    type: event,
                    name: provider.getName(e.keyCode),
                    code: e.keyCode
                };
            };
        return Rx.Observable.fromEvent(document, event)
            .filter(filter)
            .map(map);
    };

    return Rx.Observable.merge(createStream('keyup'), createStream('keydown'));
};

有趣的是，上面的代码在测试条件下(使用 Rx.TestScheduler 模拟源和调度程序)按预期工作。但在生产中，当调度程序根本没有被传递并且源是 createPress(上面)的结果时，进度流只会发出直到完成，然后再也不会发出。就好像重复完全被忽略或多余。我不知道为什么。

我在这里遗漏了什么吗？

Answer 1

您可以使用Window 。在这种情况下，我建议 WindowWithTime 。您还可以做更多有趣的事情，例如使用 Window(windowBoundaries)然后使用 Debounce 传递源代码作为边界。

source
  .windowWithTime(1500)
  .flatMap(ob => ob.reduce((acc, cur) => acc + cur, ""))

此外，由于我们的窗口是关闭的可观察对象，因此我们可以使用 Reduce累积窗口中的值并连接我们的数字。

<小时/>

现在，此变体将在 1.5 秒后关闭。相反，我们希望在最后一次按键后等待 x 秒。天真的我们可以做 source.window(source.debounce(1000)) 但现在我们订阅了我们的源两次，这是我们想要避免的事情，原因有两个。首先我们不知道订阅有什么副作用，其次我们不知道订阅的顺序会收到事件。最后一件事不是问题，因为我们使用的防抖已经在最后一次按键后增加了延迟，但仍然需要考虑。

解决办法是publish我们的来源。为了将发布保持在序列内，我们将其包装到 observable.create 中。 .

Rx.Observable.create(observer => {
    var ob = source.publish();
    return new Rx.CompositeDisposable(
      ob.window(ob.debounce(1000))
        .subscribe(observer),
      ob.connect());
}).flatMap(ob => ob.reduce((acc, cur) => acc + cur, ""))

编辑:或使用publish像这样:

source.publish(ob => ob.window(ob.debounce(1000)))
    .flatMap(ob => ob.reduce((acc, cur) => acc + cur, ""))