Merge pull request #33 from tusharmath/switch

feat(switch): add switch operator

Author: tusharmath

src/main.ts

@@ -4,7 +4,6 @@
 
 // Observer
 export {Observer} from './lib/Observer'
-
 // Operators
 import {filter as _filter} from './operators/Filter'
 import {join as _join} from './operators/Join'
@@ -13,7 +12,7 @@ import {scan as _scan} from './operators/Scan'
 import {slice as _slice} from './operators/Slice'
 import {tap as _tap} from './operators/Tap'
 import {reduce as _reduce} from './operators/Reduce'
-
+import {switchLatest as _switchLatest} from './operators/Switch'
 // Sources
 import {fromArray as _fromArray} from './sources/FromArray'
 import {interval as _interval} from './sources/Interval'
@@ -31,3 +30,4 @@ export const reduce = Curry(_reduce)
 export const fromArray = Curry(_fromArray)
 export const interval = Curry(_interval)
 export const fromDOM = Curry(_fromDOM)
+export const switchLatest = Curry(_switchLatest)

src/operators/Switch.ts

@@ -0,0 +1,72 @@
+/**
+ * Created by tushar.mathur on 16/10/16.
+ */
+import {IObservable} from '../types/core/IObservable'
+import {ISubscription} from '../types/core/ISubscription'
+import {IScheduler} from '../types/IScheduler'
+import {IObserver} from '../types/core/IObserver'
+import {CompositeSubscription} from '../lib/CompositeSubscription'
+import {Node} from '../lib/LinkedList'
+
+
+export class SwitchValueObserver<T> implements IObserver<T> {
+  constructor (private sink: IObserver<T>) {
+  }
+
+  next (val: T): void {
+    this.sink.next(val)
+  }
+
+  error (err: Error): void {
+    this.sink.error(err)
+  }
+
+  complete (): void {
+  }
+}
+
+export class SwitchObserver<T> implements IObserver<IObservable<T>> {
+  private currentSub: Node<ISubscription> | undefined = void 0
+
+  constructor (private sink: IObserver<T>,
+               private cSub: CompositeSubscription,
+               private scheduler: IScheduler) {
+  }
+
+  private removeCurrentSub () {
+    if (this.currentSub) this.cSub.remove(this.currentSub)
+  }
+
+  private setCurrentSub (subscription: ISubscription) {
+    this.removeCurrentSub()
+    this.currentSub = this.cSub.add(subscription)
+  }
+
+  next (val: IObservable<T>): void {
+    this.setCurrentSub(val.subscribe(new SwitchValueObserver(this.sink), this.scheduler))
+  }
+
+  error (err: Error): void {
+    this.sink.error(err)
+  }
+
+  complete (): void {
+    this.removeCurrentSub()
+    this.sink.complete()
+  }
+}
+
+export class SwitchLatest<T> implements IObservable<T> {
+  constructor (private source: IObservable<IObservable<T>>) {
+  }
+
+  subscribe (observer: IObserver<T>, scheduler: IScheduler): ISubscription {
+    const cSub = new CompositeSubscription()
+    cSub.add(this.source.subscribe(new SwitchObserver(observer, cSub, scheduler), scheduler))
+    return cSub
+  }
+}
+
+export function switchLatest<T> (source: IObservable<IObservable<T>>) {
+  return new SwitchLatest(source)
+}

test/test.Switch.ts

@@ -0,0 +1,41 @@
+/**
+ * Created by tushar.mathur on 16/10/16.
+ */
+import test from 'ava'
+import {TestScheduler} from '../src/testing/TestScheduler'
+import {ReactiveEvents} from '../src/testing/ReactiveEvents'
+import {switchLatest} from '../src/operators/Switch'
+
+
+test(t => {
+  const sh = TestScheduler.of()
+  const a$$ = sh.Hot([
+    ReactiveEvents.next(210, 'A0'),
+    ReactiveEvents.next(220, 'A1'),
+    ReactiveEvents.next(230, 'A2'),
+    ReactiveEvents.next(240, 'A3'),
+    ReactiveEvents.complete(250)
+  ])
+
+  const b$$ = sh.Hot([
+    ReactiveEvents.next(230, 'B0'),
+    ReactiveEvents.next(240, 'B1'),
+    ReactiveEvents.complete(250)
+  ])
+
+  const source$ = sh.Hot([
+    ReactiveEvents.next(205, a$$),
+    ReactiveEvents.next(225, b$$),
+    ReactiveEvents.complete(300)
+  ])
+  const {results} = sh.start(() => switchLatest(source$))
+  t.deepEqual(results, [
+    ReactiveEvents.next(210, 'A0'),
+    ReactiveEvents.next(220, 'A1'),
+    ReactiveEvents.next(230, 'B0'),
+    ReactiveEvents.next(240, 'B1'),
+    ReactiveEvents.complete(300)
+  ])
+  t.deepEqual(a$$.subscriptions.map(r => r.time), [205, 225])
+  t.deepEqual(b$$.subscriptions.map(r => r.time), [225, 300])
+})