Merge pull request #25 from tusharmath/skipRepeats

Skip repeats

src/operators/SkipRepeats.ts

@@ -0,0 +1,51 @@
+/**
+ * Created by niranjan on 12/10/16.
+ */
+
+import {IObservable} from '../types/core/IObservable'
+import {IObserver} from '../types/core/IObserver'
+import {ISubscription} from '../types/core/ISubscription'
+import {IScheduler} from '../types/IScheduler'
+import {IHash} from '../types/IHash'
+
+class SkipRepeatsObserver <T, H> implements IObserver<T> {
+  private hash: H | void = undefined
+  private init = true
+
+  constructor (private hasher: IHash<T, H>, private sink: IObserver<T>) {
+  }
+
+  next (val: T) {
+    const hash = this.hasher(val)
+    if (this.init) {
+      this.init = false
+      this.sink.next(val)
+      this.hash = hash
+    }
+    else if (this.hash !== hash) {
+      this.sink.next(val)
+      this.hash = hash
+    }
+  }
+
+  error (err: Error) {
+    this.sink.error(err)
+  }
+
+  complete (): void {
+    this.sink.complete()
+  }
+}
+
+export class SkipRepeatsObservable <T, H> implements IObservable <T> {
+  constructor (private hashFunction: IHash<T, H>, private source: IObservable<T>) {
+  }
+
+  subscribe (observer: IObserver<T>, scheduler: IScheduler): ISubscription {
+    return this.source.subscribe(new SkipRepeatsObserver(this.hashFunction, observer), scheduler)
+  }
+}
+
+export function skipRepeats<T, H> (hashFunction: IHash<T, H>, source: IObservable<T>): IObservable<T> {
+  return new SkipRepeatsObservable(hashFunction, source)
+}

src/types/IHash.ts

@@ -0,0 +1,7 @@
+/**
+ * Created by niranjan on 12/10/16.
+ */
+
+export interface IHash <T, H> {
+  (a: T): H
+}

test/test.SkipRepeatsObservable.ts

@@ -0,0 +1,31 @@
+/**
+ * Created by niranjan on 12/10/16.
+ */
+
+'use strict'
+
+import test from 'ava'
+import {skipRepeats} from '../src/operators/SkipRepeats'
+import {TestScheduler} from '../src/testing/TestScheduler'
+import {ReactiveEvents} from '../src/testing/ReactiveEvents'
+
+const {next, complete} = ReactiveEvents
+
+test('SkipRepeatsObservable.subscribe()', t => {
+  const sh = TestScheduler.of()
+  const $ = sh.Cold<number>([
+    next(210, 0),
+    next(215, 0),
+    next(220, 10),
+    next(230, 20),
+    next(235, 20),
+    complete(250)
+  ])
+  const {results} = sh.start(() => skipRepeats<number, number>((x: number) => x, $))
+  t.deepEqual(results, [
+    next(410, 0),
+    next(420, 10),
+    next(430, 20),
+    complete(450)
+  ])
+})