Merge pull request #154 from tusharmath/combine-new

Combine Operator

Author: tusharmath

benchmarks/bm.combine.ts

@@ -0,0 +1,26 @@
+/**
+ * Created by tushar on 08/12/16.
+ */
+import {Suite} from 'benchmark'
+import {combine} from '../src/operators/Combine'
+import {reduce} from '../src/operators/Reduce'
+import {fromArray} from '../src/sources/FromArray'
+import {array, IDeferred, run} from './lib'
+
+const a = array(1e2)
+const b = array(1e2)
+const c = array(1e2)
+const sum3 = (a: number, b: number, c: number) => a + b + c
+const sum2 = (a: number, b: number) => a + b
+
+export function bm_fromArray_combine (suite: Suite) {
+  return suite
+    .add(
+      'file -> combine(sum3, [a, b, c]) -> reduce(sum2, 0)',
+      (d: IDeferred) => run(
+        reduce(sum2, 0, combine(sum3, [fromArray(a), fromArray(b), fromArray(c)])),
+        d
+      ),
+      {defer: true}
+    )
+}

benchmarks/run.ts

@@ -2,6 +2,7 @@
  * Created by tushar.mathur on 01/10/16.
  */
 import {Suite} from 'benchmark'
+import {bm_fromArray_combine} from './bm.combine'
 import {bm_create} from './bm.create'
 import {bm_debounce} from './bm.debounce'
 import {bm_fromArray_map_reduce} from './bm.fromArray-map-reduce'
@@ -19,6 +20,7 @@ console.log('**V8:**  ', process.versions.v8)
 const suite = new Suite()
 bm_create(suite)
 bm_debounce(suite)
+bm_fromArray_combine(suite)
 bm_fromArray_map_reduce(suite)
 bm_fromArray_scan_reduce(suite)
 bm_fromArray_takeN(suite)

extra.ts

@@ -2,7 +2,5 @@
  * Created by tushar on 10/12/16.
  */
 
-export {curry} from './src/lib/Utils'
-export {EVENT} from './src/testing/Events'
-export {TestScheduler} from './src/testing/TestScheduler'
-export {marble, toMarble} from './src/testing/Marble'
+export {combine} from './src/operators/Combine'
+export {debounce} from './src/operators/Debounce'

package.json

@@ -58,5 +58,10 @@
     "ghooks": {
       "commit-msg": "validate-commit-msg"
     }
+  },
+  "ava": {
+    "files": [
+      "test/test.*.js"
+    ]
   }
 }

src/main.ts

@@ -2,8 +2,8 @@
  * Created by tushar on 17/02/17.
  */
 
+
 export {create} from './sources/Create'
-export {debounce} from './operators/Debounce'
 export {delay} from './operators/Delay'
 export {filter} from './operators/Filter'
 export {forEach} from './lib/ForEach'

src/operators/Combine.ts

@@ -0,0 +1,76 @@
+/**
+ * Created by tushar on 08/12/16.
+ */
+
+
+import {container} from '../lib/Container'
+import {IObservable} from '../lib/Observable'
+import {IObserver} from '../lib/Observer'
+import {IScheduler} from '../lib/Scheduler'
+import {CompositeSubscription, ISubscription} from '../lib/Subscription'
+import {curry} from '../lib/Utils'
+export type TSelector<T> = {(...e: Array<any>): T}
+export type TSource = Array<IObservable<any>>
+export type TResult <T> = IObservable<T>
+
+class CombineValueObserver<T> implements IObserver<T> {
+  constructor (private id: number, private sink: CombinedObserver<T>) {
+  }
+
+  next (val: T): void {
+    this.sink.onNext(val, this.id)
+  }
+
+  error (err: Error): void {
+    this.sink.onError(err)
+  }
+
+  complete (): void {
+    this.sink.onComplete(this.id)
+  }
+}
+
+class CombinedObserver<T> {
+  private collection = container(this.total)
+
+  constructor (private func: TSelector<T>, private total: number, private sink: IObserver<T>) {
+  }
+
+  onNext (value: T, id: number) {
+    this.collection.next(value, id)
+    if (this.collection.isOn()) {
+      this.sink.next(this.func.apply(null, this.collection.values))
+    }
+  }
+
+  onComplete (id: number) {
+    const hasCompleted = this.collection.complete(id)
+    if (hasCompleted) {
+      this.sink.complete()
+    }
+  }
+
+  onError (err: Error): void {
+    this.sink.error(err)
+  }
+}
+
+class CombineObservable<T> implements IObservable<T> {
+  constructor (private selector: TSelector<T>, private sources: Array<IObservable<any>>) {
+  }
+
+  subscribe (observer: IObserver<T>, scheduler: IScheduler): ISubscription {
+    const cSub = new CompositeSubscription()
+    const ob = new CombinedObserver(this.selector, this.sources.length, observer)
+    for (var i = 0; i < this.sources.length; ++i) {
+      cSub.add(this.sources[i].subscribe(new CombineValueObserver(i, ob), scheduler))
+    }
+    return cSub
+  }
+}
+
+export const combine = curry(<T> (selector: TSelector<T>, sources: IObservable<any>[]) =>
+  new CombineObservable(selector, sources)
+) as Function &
+  {<T, R> (selector: TSelector<T>, sources: TSource): TResult<R>} &
+  {<T, R> (selector: TSelector<T>): {(sources: TSource): TResult<R>}}

test.ts

@@ -0,0 +1,7 @@
+/**
+ * Created by tushar on 13/04/17.
+ */
+
+export {EVENT} from './src/testing/Events'
+export {TestScheduler} from './src/testing/TestScheduler'
+export {marble, toMarble} from './src/testing/Marble'

test/test.Combine.ts

@@ -0,0 +1,33 @@
+/**
+ * Created by tushar on 26/02/17.
+ */
+import {test} from 'ava'
+import {combine} from '../src/operators/Combine'
+import {EVENT} from '../src/testing/Events'
+import {marble} from '../src/testing/Marble'
+import {TestScheduler} from '../src/testing/TestScheduler'
+
+test(t => {
+  const SH = TestScheduler.of()
+
+  const {results} = SH.start(() => {
+    return combine(
+      (a, b, c) => a + b + c,
+      [
+        SH.Hot(marble('a-b-c-d|')),
+        SH.Hot(marble('-p-q-r-s|')),
+        SH.Hot(marble('---x-y-z--|'))
+      ]
+    )
+  })
+  t.deepEqual(results, [
+    EVENT.next(230, 'bqx'),
+    EVENT.next(240, 'cqx'),
+    EVENT.next(250, 'crx'),
+    EVENT.next(250, 'cry'),
+    EVENT.next(260, 'dry'),
+    EVENT.next(270, 'dsy'),
+    EVENT.next(270, 'dsz'),
+    EVENT.complete(300)
+  ])
+})