feat(sequenceEqual): adds sequenceEqual operator

- adds most basic tests
- documentation still required
- more tests necessary

resolves #1882

spec/operators/sequenceEqual-spec.ts

@@ -0,0 +1,167 @@
+//declare const {rxTestScheduler, time, hot, cold, asDiagram, expectObservable, expectSubscriptions, type};
+
+const booleans = { T: true, F: false };
+
+/** @test {sequenceEqual} */
+describe('Observable.prototype.sequenceEqual', () => {
+  it('should return true for two equal sequences', () => {
+    const s1 = hot('--a--^--b--c--d--e--f--g--|');
+    const s1subs =      '^                        !';
+    const s2 = hot('-----^-----b--c--d-e-f------g-|');
+    const s2subs =      '^                        !';
+    const expected =    '-------------------------(T|)';
+
+    const source = s1.sequenceEqual(s2);
+
+    expectObservable(source).toBe(expected, booleans);
+    expectSubscriptions(s1.subscriptions).toBe(s1subs);
+    expectSubscriptions(s2.subscriptions).toBe(s2subs);
+  });
+
+  it('should error if the comparor errors', () => {
+    const s1 = hot('--a--^--b-----c------d--|');
+    const s1subs =      '^            !';
+    const s2 = hot('-----^--------x---y---z-------|');
+    const s2subs =      '^            !';
+    const expected =    '-------------#';
+
+    let i = 0;
+    const source = s1.sequenceEqual(s2, (a: any, b: any) => {
+      if (++i === 2) {
+        throw new Error('shazbot');
+      }
+      return a.value === b.value;
+    });
+
+    const values = {
+      a: null,
+      b: { value: 'bees knees' },
+      c: { value: 'carpy dumb' },
+      d: { value: 'derp' },
+      x: { value: 'bees knees', foo: 'lol' },
+      y: { value: 'carpy dumb', scooby: 'doo' },
+      z: { value: 'derp', weCouldBe: 'dancin, yeah' }
+    };
+
+    expectObservable(source).toBe(expected, Object.assign({}, booleans, values), new Error('shazbot'));
+    expectSubscriptions(s1.subscriptions).toBe(s1subs);
+    expectSubscriptions(s2.subscriptions).toBe(s2subs);
+  });
+
+  it('should use the provided comparor', () => {
+    const s1 = hot('--a--^--b-----c------d--|');
+    const s1subs =      '^                        !';
+    const s2 = hot('-----^--------x---y---z-------|');
+    const s2subs =      '^                        !';
+    const expected =    '-------------------------(T|)';
+
+    const source = s1.sequenceEqual(s2, (a: any, b: any) => a.value === b.value);
+
+    const values = {
+      a: null,
+      b: { value: 'bees knees' },
+      c: { value: 'carpy dumb' },
+      d: { value: 'derp' },
+      x: { value: 'bees knees', foo: 'lol' },
+      y: { value: 'carpy dumb', scooby: 'doo' },
+      z: { value: 'derp', weCouldBe: 'dancin, yeah' }
+    };
+
+    expectObservable(source).toBe(expected, Object.assign({}, booleans, values));
+    expectSubscriptions(s1.subscriptions).toBe(s1subs);
+    expectSubscriptions(s2.subscriptions).toBe(s2subs);
+  });
+
+  it('should return false for two unequal sequences, compareTo finishing last', () => {
+    const s1 = hot('--a--^--b--c--d--e--f--g--|');
+    const s1subs =      '^                      !';
+    const s2 = hot('-----^-----b--c--d-e-f------z-|');
+    const s2subs =      '^                      !';
+    const expected =    '-----------------------(F|)';
+
+    const source = s1.sequenceEqual(s2);
+
+    expectObservable(source).toBe(expected, booleans);
+    expectSubscriptions(s1.subscriptions).toBe(s1subs);
+    expectSubscriptions(s2.subscriptions).toBe(s2subs);
+  });
+
+  it('should return false for two unequal sequences, early wrong value from source', () => {
+    const s1 = hot('--a--^--b--c---x-----------|');
+    const s1subs =      '^         !';
+    const s2 = hot('-----^--b--c--d--e--f--|');
+    const s2subs =      '^         !';
+    const expected =    '----------(F|)';
+
+    const source = s1.sequenceEqual(s2);
+
+    expectObservable(source).toBe(expected, booleans);
+    expectSubscriptions(s1.subscriptions).toBe(s1subs);
+    expectSubscriptions(s2.subscriptions).toBe(s2subs);
+  });
+
+  it('should return false when the source emits an extra value after the compareTo completes', () => {
+    const s1 = hot('--a--^--b--c--d--e--f--g--h--|');
+    const s1subs =      '^           !';
+    const s2 = hot('-----^--b--c--d-|');
+    const s2subs =      '^           !';
+    const expected =    '------------(F|)';
+
+    const source = s1.sequenceEqual(s2);
+
+    expectObservable(source).toBe(expected, booleans);
+    expectSubscriptions(s1.subscriptions).toBe(s1subs);
+    expectSubscriptions(s2.subscriptions).toBe(s2subs);
+  });
+
+  it('should return false when the compareTo emits an extra value after the source completes', () => {
+    const s1 = hot('--a--^--b--c--d-|');
+    const s1subs =      '^           !';
+    const s2 = hot('-----^--b--c--d--e--f--g--h--|');
+    const s2subs =      '^           !';
+    const expected =    '------------(F|)';
+
+    const source = s1.sequenceEqual(s2);
+
+    expectObservable(source).toBe(expected, booleans);
+    expectSubscriptions(s1.subscriptions).toBe(s1subs);
+    expectSubscriptions(s2.subscriptions).toBe(s2subs);
+  });
+
+  it('should return true for two empty observables', () => {
+    const s1 =  cold('|');
+    const s2 =  cold('|');
+    const expected = '(T|)';
+
+    const source = s1.sequenceEqual(s2);
+    expectObservable(source).toBe(expected, booleans);
+  });
+
+  it('should return false for an empty observable and an observable that emits', () => {
+    const s1 =  cold('|');
+    const s2 =  cold('---a--|');
+    const expected = '---(F|)';
+
+    const source = s1.sequenceEqual(s2);
+    expectObservable(source).toBe(expected, booleans);
+  });
+
+  it('should return compare hot and cold observables', () => {
+    const s1 = hot('---a--^---b---c---d---e---f---g---h---i---j---|');
+    const s2 = cold(     '----b---c-|');
+    const expected1 =    '------------(F|)';
+    const subs1 =        '^           !';
+    const delay =        '-------------------|';
+    const s3 = cold(                        '-f---g---h---i---j---|');
+    const expected2 =    '                   ---------------------(T|)';
+    const subs2 =        '                   ^                    !';
+
+    const test1 = s1.sequenceEqual(s2);
+    const test2 = s1.sequenceEqual(s3);
+
+    expectObservable(test1).toBe(expected1, booleans);
+    rxTestScheduler.schedule(() => expectObservable(test2).toBe(expected2, booleans), time(delay));
+    expectSubscriptions(s2.subscriptions).toBe(subs1);
+    expectSubscriptions(s3.subscriptions).toBe(subs2);
+  });
+});

src/Rx.ts

@@ -109,6 +109,7 @@ import './add/operator/retryWhen';
 import './add/operator/sample';
 import './add/operator/sampleTime';
 import './add/operator/scan';
+import './add/operator/sequenceEqual';
 import './add/operator/share';
 import './add/operator/single';
 import './add/operator/skip';

src/add/operator/sequenceEqual.ts

@@ -0,0 +1,11 @@
+
+import {Observable} from '../../Observable';
+import {sequenceEqual, SequenceEqualSignature} from '../../operator/sequenceEqual';
+
+Observable.prototype.sequenceEqual = sequenceEqual;
+
+declare module '../../Observable' {
+  interface Observable<T> {
+    sequenceEqual: SequenceEqualSignature<T>;
+  }
+}

src/operator/sequenceEqual.ts

@@ -0,0 +1,136 @@
+import {Operator} from '../Operator';
+import {Observer} from '../Observer';
+import {Observable} from '../Observable';
+import {Subscriber} from '../Subscriber';
+import {tryCatch} from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
+
+export const defaultComparor = <T>(a: T, b: T) => a === b;
+
+export interface SequenceEqualSignature<T> {
+  (compareTo: Observable<T>, comparor?: (a: T, b: T) => boolean): Observable<boolean>;
+}
+
+export function sequenceEqual<T>(compareTo: Observable<T>,
+                                 comparor: (a: T, b: T) => boolean = defaultComparor): Observable<boolean> {
+  return this.lift(new SequenceEqualOperator(compareTo, comparor));
+}
+
+export class SequenceEqualOperator<T> implements Operator<T, T> {
+  constructor(private compareTo: Observable<T>,
+              private comparor: (a: T, b: T) => boolean) {
+  }
+
+  call(subscriber: Subscriber<T>, source: any): any {
+    return source._subscribe(new SequenceEqualSubscriber(subscriber, this.compareTo, this.comparor));
+  }
+}
+
+/**
+ * We need this JSDoc comment for affecting ESDoc.
+ * @ignore
+ * @extends {Ignored}
+ */
+export class SequenceEqualSubscriber<T, R> extends Subscriber<T> {
+  private _a: T[] = [];
+  private _b: T[] = [];
+  private _oneComplete = false;
+
+  constructor(destination: Observer<R>,
+              private compareTo: Observable<T>,
+              private comparor: (a: T, b: T) => boolean) {
+    super(destination);
+    this.add(compareTo.subscribe(new SequenceEqualCompareToSubscriber(destination, this)));
+  }
+
+  protected _next(value: T): void {
+    if (this._oneComplete && this._b.length === 0) {
+      this.emit(false);
+    } else {
+      this._a.push(value);
+      this.checkValues();
+    }
+  }
+
+  public _complete(): void {
+    if (this._oneComplete) {
+      const { _a, _b, comparor } = this;
+      if (_a.length !== _b.length) {
+        this.emit(false);
+      } else {
+        const len = _a.length;
+        for (let i = 0; i < len; i++) {
+          let areEqual = false;
+          let a = _a[i];
+          let b = _b[i];
+          if (comparor) {
+            areEqual = tryCatch(comparor)(a, b);
+            if (areEqual === errorObject) {
+              this.destination.error(errorObject.e);
+              return;
+            }
+          }
+          if (!areEqual) {
+            this.emit(false);
+          }
+        }
+        this.emit(true);
+      }
+    } else {
+      this._oneComplete = true;
+    }
+  }
+
+  checkValues() {
+    const { _a, _b, comparor } = this;
+    while (_a.length > 0 && _b.length > 0) {
+      let a = _a.shift();
+      let b = _b.shift();
+      let areEqual = false;
+      if (comparor) {
+        areEqual = tryCatch(comparor)(a, b);
+        if (areEqual === errorObject) {
+          this.destination.error(errorObject.e);
+        }
+      } else {
+        areEqual = a === b;
+      }
+      if (!areEqual) {
+        this.emit(false);
+      }
+    }
+  }
+
+  emit(value: boolean) {
+    const { destination } = this;
+    destination.next(value);
+    destination.complete();
+  }
+
+  nextB(value: T) {
+    if (this._oneComplete && this._a.length === 0) {
+      this.emit(false);
+    } else {
+      this._b.push(value);
+      this.checkValues();
+    }
+  }
+}
+
+class SequenceEqualCompareToSubscriber<T, R> extends Subscriber<T> {
+  constructor(destination: Observer<R>, private parent: SequenceEqualSubscriber<T, R>) {
+    super(destination);
+  }
+
+  protected _next(value: T): void {
+    this.parent.nextB(value);
+  }
+
+  protected _error(err: any): void {
+    this.parent.error(err);
+  }
+
+  protected _complete(): void {
+    this.parent._complete();
+  }
+}