Merge pull request #12 from tusharmath/test-scheduler

Test scheduler

Author: tusharmath

benchmark.txt

@@ -24,3 +24,7 @@ Sun Oct  9 19:01:52 IST 2016
 file -> map -> reduce x 545 ops/sec ±1.09% (80 runs sampled)
 file -> takeN(0, n/10) x 255 ops/sec ±0.65% (78 runs sampled)
 file -> scan -> reduce x 82.85 ops/sec ±2.59% (74 runs sampled)
+Tue Oct 11 11:37:52 IST 2016
+file -> map -> reduce x 522 ops/sec ±1.24% (78 runs sampled)
+file -> takeN(0, n/10) x 254 ops/sec ±0.86% (80 runs sampled)
+file -> scan -> reduce x 83.45 ops/sec ±1.08% (75 runs sampled)

src/testing/ColdTestObservable.ts

@@ -0,0 +1,30 @@
+/**
+ * Created by tushar.mathur on 10/10/16.
+ */
+
+import {IEvent, EventType} from '../types/IEvent';
+import {TestObservable} from './TestObservable';
+import {ISubscriptionObserver} from '../types/core/ISubscriptionObserver';
+import {EventNext} from './ReactiveTest';
+import {IScheduler} from '../types/IScheduler';
+
+export function ColdTestObservable<T> (scheduler: IScheduler, events: Array<IEvent>) {
+  return new TestObservable((observer: ISubscriptionObserver<any>) => {
+    let closed = false
+    for (var i = 0; i < events.length && !closed; i++) {
+      const event = events[i]
+      if (event.type === EventType.next) {
+        scheduler.schedule(() => observer.next((<EventNext<any>> event).value), event.time)
+      }
+      else if (event.type === EventType.complete) {
+        scheduler.schedule(() => observer.complete(), event.time)
+      }
+    }
+    return {
+      unsubscribe () {
+        closed = true
+      },
+      closed
+    }
+  })
+}

src/testing/HotTestObservable.ts

@@ -0,0 +1,43 @@
+import {EventType, IEvent} from '../types/IEvent';
+import {TestObservable} from './TestObservable';
+import {ISubscriptionObserver} from '../types/core/ISubscriptionObserver';
+import {EventNext, EventError} from './ReactiveTest';
+import {IObserver} from '../types/core/IObserver';
+import {TestScheduler} from './TestScheduler';
+/**
+ * Created by tushar.mathur on 10/10/16.
+ */
+
+
+export function dispatchEvents<T> (event: IEvent, observers: Array<IObserver<T>>, closed: Array<boolean>) {
+  observers
+    .filter((x, i) => !closed[i])
+    .forEach(ob => {
+      if (event.type === EventType.next) return ob.next((event as EventNext<T>).value)
+      if (event.type === EventType.complete) return ob.complete()
+      if (event.type === EventType.error) return ob.error((event as EventError).value)
+    })
+}
+
+export function HotTestObservable<T> (scheduler: TestScheduler, events: Array<IEvent>) {
+  const observers: Array<IObserver<T>> = []
+  const closed: Array<boolean> = []
+  events.forEach(ev => {
+      scheduler.scheduleAbsolute(
+        () => dispatchEvents(ev, observers, closed),
+        ev.time)
+    }
+  )
+  return new TestObservable((ob: ISubscriptionObserver<T>) => {
+    const i = observers.push(ob) - 1
+    closed[i] = false
+    return {
+      unsubscribe () {
+        closed[i] = true
+      },
+      get closed () {
+        return closed[i]
+      }
+    }
+  })
+}

src/testing/TestScheduler.ts

@@ -6,12 +6,11 @@ import {ITask} from '../types/ITask';
 import {IScheduler} from '../types/IScheduler';
 import {IObservable} from '../types/core/IObservable';
 import {ISubscription} from '../types/core/ISubscription';
-import {EventNext} from './ReactiveTest';
-import {IEvent, EventType} from '../types/IEvent';
+import {IEvent} from '../types/IEvent';
 import {TestObserver} from './TestObserver';
-import {TestObservable} from './TestObservable';
-import {ISubscriptionObserver} from '../types/core/ISubscriptionObserver';
+import {ColdTestObservable} from './ColdTestObservable';
 import {ISchedulingStrategy} from '../types/ISchedulingStrategy';
+import {HotTestObservable} from './HotTestObservable';
 
 class TaskSchedule {
   constructor (public task: ITask, public time: number) {
@@ -90,8 +89,8 @@ export class TestScheduler implements IScheduler {
     this.queue = residual
   }
 
-  startScheduler<T> (f: () => IObservable<T>,
-                     timing: {start: number, stop: number} = DEFAULT_TIMING): TestObserver<T> {
+  start<T> (f: () => IObservable<T>,
+            timing: {start: number, stop: number} = DEFAULT_TIMING): TestObserver<T> {
     var subscription: ISubscription
     var resultsObserver = new TestObserver(this);
     this.scheduleAbsolute(() => subscription = f().subscribe(resultsObserver, this), timing.start)
@@ -102,25 +101,12 @@ export class TestScheduler implements IScheduler {
     return resultsObserver
   }
 
-  createColdObservable <T> (events: Array<IEvent>): IObservable<T> {
-    return new TestObservable((observer: ISubscriptionObserver<any>) => {
-      let closed = false
-      for (var i = 0; i < events.length && !closed; i++) {
-        const event = events[i]
-        if (event.type === EventType.next) {
-          this.schedule(() => observer.next((<EventNext<any>> event).value), event.time)
-        }
-        else if (event.type === EventType.complete) {
-          this.schedule(() => observer.complete(), event.time)
-        }
-      }
-      return {
-        unsubscribe () {
-          closed = true
-        },
-        closed
-      }
-    })
+  Cold <T> (events: Array<IEvent>) {
+    return ColdTestObservable(this, events)
+  }
+
+  Hot <T> (events: Array<IEvent>) {
+    return HotTestObservable(this, events)
   }
 
   static of () {

src/types/IScheduler.ts

@@ -7,7 +7,7 @@ import {ITask} from './ITask';
 import {ISchedulingStrategy} from './ISchedulingStrategy';
 
 export interface IScheduler {
-  schedule(task: ITask, time: number): ISubscription
+  schedule(task: ITask, relativeTime: number): ISubscription
   scheduleASAP(task: ITask): ISubscription
   scheduleNow(task: ITask): ISubscription
   scheduleRepeatedly(task: ITask, interval: number): ISubscription

test/test.FromArray.ts

@@ -14,7 +14,7 @@ const {next, error} = ReactiveTest
 test(t => {
   const sh = TestScheduler.of()
   const testFunction = (x: any) => x === 2 ? x.do() : x * 100
-  const {results} = sh.startScheduler(() => map(testFunction, fromArray([1, 2, 3])))
+  const {results} = sh.start(() => map(testFunction, fromArray([1, 2, 3])))
   t.deepEqual(results, [
     next(201, 100),
     error(201, new TypeError())

test/test.IntervalObservable.ts

@@ -13,7 +13,7 @@ const {next, error} = ReactiveTest
 
 test('subscribe()', t => {
   const sh = TestScheduler.of()
-  const {results} = sh.startScheduler<number>(() => interval(200))
+  const {results} = sh.start<number>(() => interval(200))
   t.deepEqual(results, [
     next(400, 0),
     next(600, 1),

test/test.JoinObservable.ts

@@ -10,24 +10,24 @@ const {next, complete} = ReactiveTest
 
 test('subscribe()', t => {
   const sh = TestScheduler.of()
-  const sa$$ = sh.createColdObservable([
+  const sa$$ = sh.Cold([
     next(10, 'A0'),
     next(20, 'A1'),
     next(30, 'A2'),
     complete(40)
   ])
-  const sb$$ = sh.createColdObservable([
+  const sb$$ = sh.Cold([
     next(10, 'B0'),
     next(20, 'B1'),
     next(30, 'B2'),
     complete(40)
   ])
-  const s$$ = sh.createColdObservable([
+  const s$$ = sh.Cold([
     next(10, sa$$),
     next(20, sb$$),
     complete(100)
   ])
-  const {results} = sh.startScheduler<number>(() => join(s$$))
+  const {results} = sh.start<number>(() => join(s$$))
 
   t.deepEqual(results, [
     next(220, 'A0'),
@@ -39,3 +39,39 @@ test('subscribe()', t => {
     complete(300)
   ])
 })
+
+
+test('subscribe():hot', t => {
+  const sh = TestScheduler.of()
+
+  // start(210)
+  const sa$$ = sh.Hot([
+    next(211, 'A0'),
+    next(220, 'A1'),
+    next(230, 'A2'),
+    complete(240)
+  ])
+
+  // start(220)
+  const sb$$ = sh.Hot([
+    next(205, 'B0'),
+    next(215, 'B1'),
+    next(225, 'B2'),
+    complete(340)
+  ])
+
+  // start(210)
+  const s$$ = sh.Hot([
+    next(210, sa$$),
+    next(220, sb$$),
+    complete(300)
+  ])
+  const {results} = sh.start<number>(() => join(s$$))
+  t.deepEqual(results, [
+    next(211, 'A0'),
+    next(220, 'A1'),
+    next(225, 'B2'),
+    next(230, 'A2'),
+    complete(340)
+  ])
+})

test/test.MapObservable.ts

@@ -13,17 +13,35 @@ const {next, complete} = ReactiveTest
 
 test('MapObservable.subscribe()', t => {
   const sh = TestScheduler.of()
-  const $ = sh.createColdObservable<number>([
+  const $ = sh.Cold<number>([
     next(210, 0),
     next(220, 10),
     next(230, 20),
     complete(250)
   ])
-  const {results} = sh.startScheduler(() => map<number>((x: number) => x + 1, $))
+  const {results} = sh.start(() => map<number>((x: number) => x + 1, $))
   t.deepEqual(results, [
     next(410, 1),
     next(420, 11),
     next(430, 21),
     complete(450)
   ])
 })
+
+test('MapObservable.subscribe():HOT', t => {
+  const sh = TestScheduler.of()
+  const $ = sh.Hot<number>([
+    next(100, -10),
+    next(210, 0),
+    next(220, 10),
+    next(230, 20),
+    complete(250)
+  ])
+  const {results} = sh.start(() => map<number>((x: number) => x + 1, $))
+  t.deepEqual(results, [
+    next(210, 1),
+    next(220, 11),
+    next(230, 21),
+    complete(250)
+  ])
+})

test/test.ScanObservable.ts

@@ -11,15 +11,15 @@ const {next, complete} = ReactiveTest
 
 test('ScanObservable.subscribe()', t => {
   const sh = TestScheduler.of()
-  const $ = sh.createColdObservable<number>([
+  const $ = sh.Cold<number>([
     next(210, 0),
     next(220, 1),
     next(230, 2),
     next(240, 3),
     next(250, 4),
     complete(250)
   ])
-  const {results} = sh.startScheduler(() => scan((a, b) => a + b, 0, $))
+  const {results} = sh.start(() => scan((a, b) => a + b, 0, $))
   t.deepEqual(results, [
     next(410, 0),
     next(420, 1),

test/test.SliceObservable.ts

@@ -12,15 +12,15 @@ import {ReactiveTest} from '../src/testing/ReactiveTest';
 const {next, complete} = ReactiveTest
 test('takeN(0, 3)', t => {
   const sh = TestScheduler.of()
-  const ob$ = sh.createColdObservable([
+  const ob$ = sh.Cold([
     next(0, 1),
     next(10, 2),
     next(20, 3),
     next(30, 4),
     next(40, 5),
     complete(50)
   ])
-  const {results} = sh.startScheduler(() => slice(0, 3, ob$))
+  const {results} = sh.start(() => slice(0, 3, ob$))
   t.deepEqual(results, [
     next(200, 1),
     next(210, 2),
@@ -31,7 +31,7 @@ test('takeN(0, 3)', t => {
 
 test('takeN(0, Infinity)', t => {
   const sh = TestScheduler.of()
-  const ob$ = sh.createColdObservable([
+  const ob$ = sh.Cold([
     next(0, 1),
     next(10, 2),
     next(20, 3),
@@ -40,7 +40,7 @@ test('takeN(0, Infinity)', t => {
     complete(50),
     next(60, 6)
   ])
-  const {results} = sh.startScheduler(() => slice(0, Infinity, ob$))
+  const {results} = sh.start(() => slice(0, Infinity, ob$))
   t.deepEqual(results, [
     next(200, 1),
     next(210, 2),
@@ -53,15 +53,15 @@ test('takeN(0, Infinity)', t => {
 
 test('takeN(1, 3)', t => {
   const sh = TestScheduler.of()
-  const ob$ = sh.createColdObservable([
+  const ob$ = sh.Cold([
     next(0, 1),
     next(10, 2),
     next(20, 3),
     next(30, 4),
     next(40, 5),
     complete(50)
   ])
-  const {results} = sh.startScheduler(() => slice(1, 3, ob$))
+  const {results} = sh.start(() => slice(1, 3, ob$))
   t.deepEqual(results, [
     next(210, 2),
     next(220, 3),