perf(ScalarObservable): add fast-path for mapping scalar observables

ReactiveX · Oct 14, 2015 · 7b0d3dc · 7b0d3dc
1 parent acf56ec
commit 7b0d3dc
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 3 deletions.
diff --git a/spec/operators/retry-spec.js b/spec/operators/retry-spec.js
@@ -6,7 +6,10 @@ describe('Observable.prototype.retry()', function () {
   it('should retry a number of times, without error, then complete', function (done) {
     var errors = 0;
     var retries = 2;
-    Observable.of(42)
+    Observable.create(function (observer) {
+      observer.next(42);
+      observer.complete();
+    })
       .map(function (x) {
         if ((errors += 1) < retries) {
           throw 'bad';
@@ -27,7 +30,10 @@ describe('Observable.prototype.retry()', function () {
   it('should retry a number of times, then call error handler', function (done) {
     var errors = 0;
     var retries = 2;
-    Observable.of(42)
+    Observable.create(function (observer) {
+      observer.next(42);
+      observer.complete();
+    })
       .map(function (x) {
         if ((errors += 1) < retries) {
           throw 'bad';
@@ -50,7 +56,10 @@ describe('Observable.prototype.retry()', function () {
   it('should retry until successful completion', function (done) {
     var errors = 0;
     var retries = 10;
-    Observable.of(42)
+    Observable.create(function (observer) {
+      observer.next(42);
+      observer.complete();
+    })
       .map(function (x) {
         if ((errors += 1) < retries) {
           throw 'bad';

diff --git a/src/observables/ScalarObservable.ts b/src/observables/ScalarObservable.ts
@@ -1,5 +1,9 @@
 import Scheduler from '../Scheduler';
 import Observable from '../Observable';
+import tryCatch from '../util/tryCatch';
+import { errorObject } from '../util/errorObject';
+import ErrorObservable from './ErrorObservable';
+import EmptyObservable from './EmptyObservable';
 
 export default class ScalarObservable<T> extends Observable<T> {
 
@@ -49,4 +53,13 @@ export default class ScalarObservable<T> extends Observable<T> {
       }
     }
   }
+
+  map<R>(project: (x: T, ix?: number) => R, thisArg?: any): Observable<R> {
+    let result = tryCatch(project).call(thisArg || this, this.value, 0);
+    if (result === errorObject) {
+      return new ErrorObservable(errorObject.e);
+    } else {
+      return new ScalarObservable(project.call(thisArg || this, this.value, 0));
+    }
+  }
 }