Merge pull request #127 from tusharmath/perf-updates

perf(scheduler): optimize constructor of tasks

package.json

@@ -9,17 +9,17 @@
     "cleanup": "find ./src -type f -name '*.js' -delete && find ./src -type f -name '*.map' -delete",
     "coverage": "nyc npm test && nyc report --reporter=text-lcov | coveralls",
     "hydra": "node --trace-hydrogen --trace-phase=Z --trace-deopt --code-comments --hydrogen-track-positions --redirect-code-traces --redirect-code-traces-to=code.asm .dist/benchmarks/run",
-    "prepublish": "tsc -d && npm run build && npm run build-min",
+    "prepublish": "tsc -d && npm run build && npm run build:min",
     "rfc": "node chore/rfc",
     "semantic-release": "semantic-release pre && npm publish && semantic-release post",
     "test": "tsc -d && ava .dist/test",
     "test:watch": "ava .dist/test --watch --source .dist",
-    "build": "npm run build-es && npm run build-cjs",
-    "build-min": "npm run build-es-min && npm run build-cjs-min",
-    "build-es": "rollup -c",
-    "build-es-min": "babel .dist/main-es.js > .dist/main-es.min.js",
-    "build-cjs": "browserify .dist/src/main.js > .dist/main-cjs.js",
-    "build-cjs-min": "babel .dist/main-cjs.js > .dist/main-cjs.min.js"
+    "build": "npm run build:es && npm run build:cjs",
+    "build:min": "npm run build:es:min && npm run build:cjs:min",
+    "build:es": "rollup -c",
+    "build:es:min": "babel .dist/main-es.js > .dist/main-es.min.js",
+    "build:cjs": "browserify .dist/src/main.js > .dist/main-cjs.js",
+    "build:cjs:min": "babel .dist/main-cjs.js > .dist/main-cjs.min.js"
   },
   "author": "",
   "license": "ISC",

src/lib/DefaultScheduler.ts

@@ -6,47 +6,27 @@ import {Subscription} from '../types/core/Subscription'
 import {ITask} from '../types/ITask'
 import {IScheduledTask} from '../types/IScheduledTask'
 
+type RicOptions = {timeout: number}
+declare function requestIdleCallback (fn: () => void, options?: RicOptions): number
+declare function cancelIdleCallback (id: number): void
+
 function run (task: IScheduledTask) {
   return task.run()
 }
-
-
 class AnimationFrame implements IScheduledTask {
   closed = false
   private id: number
 
   constructor (private task: ITask) {
   }
 
-  run () {
-    this.id = requestAnimationFrame(() => {
-      this.closed = true
-      this.task()
-    })
-    return this
-  }
-
-  unsubscribe (): void {
-    if (this.closed) return
-    cancelAnimationFrame(this.id)
+  onFrame () {
     this.closed = true
-  }
-}
-
-class AnimationFrames implements IScheduledTask {
-  closed = false
-  private id: number
-
-  constructor (private task: ITask) {
-  }
-
-  onFrame = () => {
     this.task()
-    if (!this.closed) this.run()
   }
 
   run () {
-    this.id = requestAnimationFrame(this.onFrame)
+    this.id = requestAnimationFrame(this.onFrame.bind(this))
     return this
   }
 
@@ -56,7 +36,6 @@ class AnimationFrames implements IScheduledTask {
     this.closed = true
   }
 }
-
 class Interval implements IScheduledTask {
   closed = false
   private id: any
@@ -66,7 +45,7 @@ class Interval implements IScheduledTask {
   }
 
   run () {
-    this.id = setInterval(() => this.task(), this.interval)
+    this.id = setInterval(this.task, this.interval)
     return this
   }
 
@@ -75,30 +54,78 @@ class Interval implements IScheduledTask {
     this.closed = true
   }
 }
-
 class Timeout implements IScheduledTask {
   closed = false
   private timer: any
 
   constructor (private task: ITask, private timeout: number) {
   }
 
+  private onTimeout () {
+    this.task()
+    this.closed = true
+  }
+
   run () {
-    this.timer = setTimeout(() => {
+    this.timer = setTimeout(this.onTimeout.bind(this), this.timeout)
+    return this
+  }
+
+  unsubscribe (): void {
+    if (this.closed === false) {
+      clearTimeout(this.timer)
       this.closed = true
-      this.task()
-    }, this.timeout)
+    }
+  }
+}
+class RequestIdleCallback implements IScheduledTask {
+  closed: boolean
+  private id: number
+
+  constructor (private task: ITask, private options?: RicOptions) {
+  }
+
+  run (): IScheduledTask {
+    this.id = requestIdleCallback(this.task, this.options)
     return this
   }
 
   unsubscribe (): void {
     if (this.closed) return
-    clearTimeout(this.timer)
-    this.closed = true
+    this.closed = false
+    cancelIdleCallback(this.id)
   }
 }
+class NextTick implements IScheduledTask {
+  closed: boolean
+
+  constructor (private task: ITask) {
+  }
+
+  onTick (i: NextTick) {
+    if (i.closed) return
+    i.task()
+  }
 
+  run (): IScheduledTask {
+    process.nextTick(this.onTick, this)
+    return this
+  }
+
+  unsubscribe (): void {
+    if (this.closed) return
+    this.closed = false
+  }
+}
 class DefaultScheduler implements Scheduler {
+  requestIdleCallback (task: ITask, options?: RicOptions): Subscription {
+    return run(new RequestIdleCallback(task, options))
+  }
+
+  nextTick (task: ITask): Subscription {
+    return run(new NextTick(task))
+  }
+
   setInterval (task: ITask, interval: number): Subscription {
     return run(new Interval(task, interval))
   }
@@ -111,13 +138,8 @@ class DefaultScheduler implements Scheduler {
     return run(new AnimationFrame(task))
   }
 
-  requestAnimationFrames (task: ITask): Subscription {
-    return run(new AnimationFrames(task))
-  }
-
   now (): number {
     return Date.now()
   }
 }
-
 export const createScheduler = (): Scheduler => new DefaultScheduler()

src/lib/ScheduleAsap.ts

@@ -0,0 +1,23 @@
+import {Scheduler} from '../types/Scheduler'
+import {ITask} from '../types/ITask'
+import {Subscription} from '../types/core/Subscription'
+/**
+ * Created by tushar on 05/02/17.
+ */
+
+interface Global {
+  requestIdleCallback?: Function
+  process?: {nextTick: Function}
+  setTimeout: Function
+}
+
+function getGlobal (): Global {
+  return typeof window === 'object' ? window : global
+}
+
+export function asap (sh: Scheduler, task: ITask): Subscription {
+  const global = getGlobal()
+  if (global.requestIdleCallback) return sh.requestIdleCallback(task)
+  if (global.process) return sh.nextTick(task)
+  return sh.setTimeout(task, 1)
+}

src/operators/Delay.ts

@@ -7,6 +7,7 @@ import {Scheduler} from '../types/Scheduler'
 import {Subscription} from '../types/core/Subscription'
 import {safeObserver} from '../lib/SafeObserver'
 import {CompositeSubscription} from '../lib/CompositeSubscription'
+import {Curry} from '../lib/Curry'
 
 class DelayObserver<T> implements Observer<T> {
   constructor (private timeout: number,
@@ -15,10 +16,6 @@ class DelayObserver<T> implements Observer<T> {
                private cSub: CompositeSubscription) {
   }
 
-  private completeDelayed = () => {
-    this.sink.complete()
-  }
-
   next (val: T): void {
     const node = this.cSub.add(this.scheduler.setTimeout(() => {
       this.sink.next(val)
@@ -31,7 +28,7 @@ class DelayObserver<T> implements Observer<T> {
   }
 
   complete (): void {
-    this.cSub.add(this.scheduler.setTimeout(this.completeDelayed, this.timeout))
+    this.cSub.add(this.scheduler.setTimeout(this.sink.complete.bind(this.sink), this.timeout))
   }
 }
 
@@ -47,4 +44,6 @@ class DelayObservable<T> implements Observable<T> {
   }
 }
 
-export const delay = <T> (timeout: number, source: Observable<T>): Observable<T> => new DelayObservable(timeout, source)
+export const delay = Curry(<T> (timeout: number, source: Observable<T>): Observable<T> => new DelayObservable(timeout, source)) as Function &
+  {<T> (timeout: number, source: Observable<T>): Observable<T>} &
+  {<T> (timeout: number): {(source: Observable<T>): Observable<T>}}

src/operators/Reduce.ts

@@ -10,7 +10,7 @@ import {Scheduler} from '../types/Scheduler'
 import {Curry} from '../lib/Curry'
 
 
-export type TReducer<T, R> = {(previousValue: R, currentValue: T): R}
+export type TReducer<T, R> = {(memory: R, current: T): R}
 export type TSeed<R> = R
 export type TSource<T> = Observable<T>
 export type TResult<R> = Observable<R>

src/operators/Scan.ts

@@ -1,15 +1,13 @@
 /**
  * Created by tushar.mathur on 09/10/16.
  */
-
-
 import {Observable} from '../types/core/Observable'
 import {Observer} from '../types/core/Observer'
 import {Scheduler} from '../types/Scheduler'
 import {Subscription} from '../types/core/Subscription'
 import {Curry} from '../lib/Curry'
 
-export type TReducer <T, R> = (current: T, memory: R) => R
+export type TReducer <T, R> = (memory: R, current: T) => R
 export type TSeed <R> = R
 export type TSource <T> = Observable<T>
 export type TResult <R> = Observable<R>
@@ -22,7 +20,7 @@ class ScanObserver<T, V> implements Observer<T> {
   }
 
   next (val: T): void {
-    this.value = this.reducer(val, this.value)
+    this.value = this.reducer(this.value, val)
     this.sink.next(this.value)
   }
 
@@ -52,4 +50,4 @@ export const scan = Curry(function <T, V> (reducer: TReducer<T, V>, value: V, so
   {<T, R>(reducer: TReducer<T, R>, seed: TSeed<R>, source: TSource<T>): TResult<R>} &
   {<T, R>(reducer: TReducer<T, R>): {(seed: TSeed<R>, source: TSource<T>): TResult<R>}} &
   {<T, R>(reducer: TReducer<T, R>, seed: TSeed<R>): {(source: TSource<T>): TResult<R>}} &
-  {<T, R>(reducer: TReducer<T, R>): { (seed: TSeed<R>): { (source: TSource<T>): TResult<R> } } }
+  {<T, R>(reducer: TReducer<T, R>): {(seed: TSeed<R>): {(source: TSource<T>): TResult<R>}}}

src/operators/Switch.ts

@@ -27,10 +27,12 @@ class SwitchValueObserver<T> implements Observer<T> {
 
 class SwitchObserver<T> implements Observer<Observable<T>> {
   private currentSub: LinkedListNode<Subscription> | undefined = void 0
+  private sink: SwitchValueObserver<T>
 
-  constructor (private sink: Observer<T>,
+  constructor (private mainSink: Observer<T>,
                private cSub: CompositeSubscription,
                private scheduler: Scheduler) {
+    this.sink = new SwitchValueObserver(mainSink)
   }
 
   private removeCurrentSub () {
@@ -43,7 +45,7 @@ class SwitchObserver<T> implements Observer<Observable<T>> {
   }
 
   next (val: Observable<T>): void {
-    this.setCurrentSub(val.subscribe(new SwitchValueObserver(this.sink), this.scheduler))
+    this.setCurrentSub(val.subscribe(this.sink, this.scheduler))
   }
 
   error (err: Error): void {
@@ -52,7 +54,7 @@ class SwitchObserver<T> implements Observer<Observable<T>> {
 
   complete (): void {
     this.removeCurrentSub()
-    this.sink.complete()
+    this.mainSink.complete()
   }
 }