识别和准确的 RxJS 中的一系列事件答案

【问题标题】：Recognizing and exact series of events in RxJS识别和准确的 RxJS 中的一系列事件
【发布时间】：2016-01-20 02:11:19
【问题描述】：

使用 RxJS，如果用户键入“a”然后“b”然后“c”，我想触发一个事件。

如果他们输入 'a' 然后 'b' 然后 'z' 然后 'c'，我不希望触发事件。

这是我迄今为止所做的工作的codepen（在 TypeScript 中）。

class App1 {
    private divStream: HTMLElement;
    private divResult: HTMLElement;

    constructor(divStream: HTMLElement, divResult: HTMLElement) {
            this.divStream = divStream;
            this.divResult = divResult;
    }

    start() {
        var filterByCharacter = (expectedCharater) => {
            return (char) => { return char === expectedCharater; };
        };

        var values = ['a', 'b', 'b', 'c', 'b'];

        var obChars = Rx.Observable.fromArray(values);
        obChars.subscribe((k) => {
            divStream.innerHTML += "<div style='color: blue'>" + ":: " + k + " ::" + "</div>";
                },
            (err) => {
                divStream.innerHTML += "<div style='background-color: blue' > " + 'Error: ' + err + " </div>";
            },
            () => {
                divStream.innerHTML += "<div style='background-color: blue'>" + ":: finished ::" + "</div>";
            }
        );

        function log(text: string) {
            divResult.innerHTML += "<div style='color: green'>" + text + "</div>";
        }
        var obA: Rx.Observable<string> = obChars.filter(filterByCharacter('a'));
        var obB: Rx.Observable<string> = obChars.filter(filterByCharacter('b'));
        var obC: Rx.Observable<string> = obChars.filter(filterByCharacter('c'));

        let aSteps: Rx.Observable<any>[] = [];
        aSteps.push(obA.take(1).do(() => { log("a"); }).ignoreElements());
        aSteps.push(obB.take(1).do(() => { log("b"); }).ignoreElements());
        aSteps.push(obC.take(1).do(() => { log("c"); }));
        let steps: Rx.Observable<any> = Rx.Observable.concat<any>(aSteps);
        var source = steps
            .takeUntil(Rx.Observable.timer(100 * values.length));

              var subscription = source.subscribe(
            function (x) {
                log("Next: " + x);
            },
            function (err) {
                divResult.innerHTML += "<div style='background-color: green'>Error: " + err + "</div>";
            },
            function () {
                divResult.innerHTML += "<div style='background-color: green' > " + 'Completed' + "</div>";
            });
    }

    stop() {
        clearTimeout(this.timerToken);
    }

}

window.onload = () => {
    var app = new App1(document.getElementById('divStream'), document.getElementById('divResult'));
    app.start();
};

【问题讨论】：

仅供参考：我看到了实现此stackoverflow.com/questions/25833495/… 的“减少”方法。我正在寻找更多 RxJS 原生的东西。
另一条评论：当我说我正在寻找更多 RxJS 原生的东西时，我的意思是我正在寻找一个声明性的答案，而不是一个命令性的答案。

标签： javascript reactive-programming rxjs

【解决方案1】：

这似乎有效，它使用simple state machine 并且可以泛化以识别任何基本的正则表达式。这里识别的正则表达式是 *abc* :

function noop () {}

var keyUp$ = 
  Rx.Observable.fromEvent(ta_input, 'keyup')
               .map(function(ev){return ev.keyCode});

var stateMachine$ = keyUp$
  .scan(function (state, keyCode) {
    if (String.fromCharCode(keyCode) === state.password[state.index]) {
      state.index++;
      if (state.index === state.password.length) {
        state.found = true;
      }
    } 
    else {
      state.index = 0;
      state.found = false;
    }
    return state;
}, {password : 'ABC', index : 0, found: false})
  .filter(function (state){return state.found})
  .take(1)

stateMachine$.subscribe(noop)

要检查它，请运行 jsffidle，然后在 textarea 中键入 abc。出于某种原因，您的密码必须大写，但该部分应该很容易修复。当检测到 abc 子字符串时，stateMachine$ observable 会发出 found 并完成。

【讨论】：

感谢您的回答。虽然这是一个有效的解决方案，但我们 RxJS 的精神不是使用状态或命令式技术，至少我是这样解释的。
这里的状态或命令式技术是什么意思？在我看来，这似乎是标准和“原生”（这甚至意味着什么）RxJS。 RxJS 通过操作符组合并由观察者订阅的源来构建流。运算符通常采用选择器功能。这些函数是命令式程序代码，无法绕过它，或者停止用 javascript 编码。优选的是输出流仅依赖于输入流和输入数据的样式。这与使用纯函数与执行副作用的函数相同。
因此，建议使用 RxJS 避免副作用或将其本地化，例如在 do 运算符中。这段代码没有副作用，它是纯函数式的，因为输出流是输入流的纯函数。即在两个不同的时间，您将给定的点击数据流传递给该函数，它将输出完全相同的内容。将stateMachine 替换为deterministicFiniteAutomaton 或exactStringMatcher，并将state 替换为step。它可能会帮助您看到这里没有状态，只有标准数据流，没有副作用。
感谢您的澄清，当然，您对我的“状态”论点是正确的。至于我对使用命令式技术的反对意见，我也许天真地希望有一个看起来更像keyUp.assertNext(function(v) { return v == 'a';}).assertNext(function(v) { return v == 'b';})..assertNext(function(v) { return v == 'c';}); 的解决方案。其中assertNext 是我的一个名义运算符，它在 true 上“接受”一个事件，在 false 上抛出一个错误。如果没有更多与此相符的内容，我会将您的答案标记为正确。
我分叉了你的 jsFiddle，制作了一些模组来让“aabc”成功完成。 jsfiddle.net/h0fab5z0

【解决方案2】：

我相信你正在寻找bufferWithCount。

查看来自 RxJSNext 的 spec（下一个重命名为 bufferCount）：

it('should emit full buffer then last partial buffer if source completes', function () {
  var e1 =   hot('--a^-b--c--d--e--|');
  var e1subs =      '^             !';
  var expected =    '--------y-----(z|)';

  expectObservable(e1.bufferCount(3)).toBe(expected, {
                                             y: ['b', 'c', 'd'],
                                             z: ['e']
                                           });
  expectSubscriptions(e1.subscriptions).toBe(e1subs);
 });

【讨论】：