refactor(every): perf improvements for scalar and array observables

perf/micro/immediate-scheduler/operators/every-predicate-array.js

@@ -0,0 +1,24 @@
+var RxOld = require("rx");
+var RxNew = require("../../../../index");
+
+module.exports = function (suite) {
+
+  var predicate = function(x) {
+    return x === 'hi';
+  }
+
+  var oldEveryPredicateArgs = RxOld.Observable.fromArray([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], RxOld.Scheduler.immediate).every(predicate);
+  var newEveryPredicateArgs = RxNew.Observable.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).every(predicate);
+
+  return suite
+      .add('old static array observable every(predicate) with immediate scheduler', function () {
+          oldEveryPredicateArgs.subscribe(_next, _error, _complete);
+      })
+      .add('new static array observable every(predicate) with immediate scheduler', function () {
+          newEveryPredicateArgs.subscribe(_next, _error, _complete);
+      });
+
+  function _next(x) { }
+  function _error(e){ }
+  function _complete(){ }
+};

perf/micro/immediate-scheduler/operators/every-predicate-scalar.js

@@ -0,0 +1,24 @@
+var RxOld = require("rx");
+var RxNew = require("../../../../index");
+
+module.exports = function (suite) {
+
+  var predicate = function(x) {
+    return x === 'hi';
+  }
+
+  var oldEveryPredicateArgs = RxOld.Observable.of('hi', RxOld.Scheduler.immediate).every(predicate);
+  var newEveryPredicateArgs = RxNew.Observable.of('hi').every(predicate);
+
+  return suite
+      .add('old scalar observable with every(predicate) with immediate scheduler', function () {
+          oldEveryPredicateArgs.subscribe(_next, _error, _complete);
+      })
+      .add('new scalar observable with every(predicate) with immediate scheduler', function () {
+          newEveryPredicateArgs.subscribe(_next, _error, _complete);
+      });
+
+  function _next(x) { }
+  function _error(e){ }
+  function _complete(){ }
+};

src/observables/ArrayObservable.ts

@@ -50,7 +50,7 @@ export default class ArrayObservable<T> extends Observable<T> {
   // value used if Array has one value and _isScalar
   value: any;
 
-  constructor(private array: T[], private scheduler?: Scheduler) {
+  constructor(public array: T[], public scheduler?: Scheduler) {
     super();
     if(!scheduler && array.length === 1) {
       this._isScalar = true;

src/observables/PromiseObservable.ts

@@ -13,7 +13,7 @@ export default class PromiseObservable<T> extends Observable<T> {
     return new PromiseObservable(promise, scheduler);
   }
 
-  constructor(private promise: Promise<T>, private scheduler: Scheduler) {
+  constructor(private promise: Promise<T>, public scheduler: Scheduler) {
     super();
   }
 

src/observables/ScalarObservable.ts

@@ -29,7 +29,7 @@ export default class ScalarObservable<T> extends Observable<T> {
 
   _isScalar: boolean = true;
 
-  constructor(public value: T, private scheduler?: Scheduler) {
+  constructor(public value: T, public scheduler?: Scheduler) {
     super();
   }
 

src/operators/every.ts

@@ -1,14 +1,38 @@
 import Operator from '../Operator';
 import Observer from '../Observer';
 import Observable from '../Observable';
+import ScalarObservable from '../observables/ScalarObservable';
+import ArrayObservable from '../observables/ArrayObservable';
+import ErrorObservable from '../observables/ErrorObservable';
 import Subscriber from '../Subscriber';
-
+import immediate from '../schedulers/immediate';
 import tryCatch from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
 import bindCallback from '../util/bindCallback';
 
 export default function every<T>(predicate: (value: T, index: number, source: Observable<T>) => boolean, thisArg?: any): Observable<T>{
-  return this.lift(new EveryOperator(predicate, thisArg, this));
+  const source = this;
+  let result;
+
+  if(source._isScalar) {
+    result = tryCatch(predicate)(source.value, 0, source);
+    if(result === errorObject) {
+      return new ErrorObservable(errorObject.e, source.scheduler);
+    } else {
+      return new ScalarObservable(result, source.scheduler);
+    }
+  }
+
+  if(source instanceof ArrayObservable) {
+    const array = (<ArrayObservable<T>>source).array;
+    let result = tryCatch((array, predicate) => array.every(<any>predicate))(array, predicate);
+    if(result === errorObject) {
+      return new ErrorObservable(errorObject.e, source.scheduler);
+    } else {
+      return new ScalarObservable(result, source.scheduler);
+    }
+  }
+  return source.lift(new EveryOperator(predicate, thisArg, source));
 }
 
 class EveryOperator<T, R> implements Operator<T, R> {