feat(pairwise): implement extra operator pairwise()

src/extra/pairwise.ts

@@ -0,0 +1,42 @@
+import {Operator} from '../Operator';
+import {Stream} from '../Stream';
+
+class PairwiseOperator<T> implements Operator<T, [T, T]> {
+  private val: T = null;
+  private has: boolean = false;
+  private out: Stream<[T, T]> = null;
+
+  constructor(public ins: Stream<T>) {
+  }
+
+  _start(out: Stream<[T, T]>): void {
+    this.out = out;
+    this.ins._add(this);
+  }
+
+  _stop(): void {
+    this.ins._remove(this);
+  }
+
+  _n(t: T) {
+    if (this.has) {
+      this.out._n([this.val, t]);
+    }
+    this.val = t;
+    this.has = true;
+  }
+
+  _e(err: any) {
+    this.out._e(err);
+  }
+
+  _c() {
+    this.out._c();
+  }
+}
+
+export default function pairwise<T>(): (ins: Stream<T>) => Stream<[T, T]> {
+  return function pairwiseOperator(ins: Stream<T>): Stream<[T, T]> {
+    return new Stream<[T, T]>(new PairwiseOperator(ins));
+  };
+}

tests/extra/pairwise.ts

@@ -0,0 +1,30 @@
+import xs from '../../src/index';
+import pairwise from '../../src/extra/pairwise';
+import * as assert from 'assert';
+
+describe('pairwise (extra)', () => {
+  it('should group consecutive pairs as arrays', (done) => {
+    const stream = xs.of(1, 2, 3, 4, 5, 6).compose(pairwise());
+    const expected = [
+      [1, 2],
+      [2, 3],
+      [3, 4],
+      [4, 5],
+      [5, 6],
+    ];
+
+    stream.addListener({
+      next: (x: [number, number]) => {
+        const e = expected.shift();
+        assert.equal(x.length, e.length);
+        assert.equal(x[0], e[0]);
+        assert.equal(x[1], e[1]);
+      },
+      error: (err: any) => done(err),
+      complete: () => {
+        assert.equal(expected.length, 0);
+        done();
+      },
+    });
+  });
+});