Map transducing

perf/micro/immediate-scheduler/operators/map-mergeAll.js

@@ -0,0 +1,24 @@
+var RxOld = require('rx');
+var RxNew = require('../../../../index');
+
+module.exports = function (suite) {
+  function toObservableOld(x) {
+    return RxOld.Observable.range(0, 3);
+  }
+  function toObservableNew(x) {
+    return RxNew.Observable.range(0, 3);
+  }
+  var _old = RxOld.Observable.range(0, 50, RxOld.Scheduler.immediate).map(toObservableOld).mergeAll();
+  var _new = RxNew.Observable.range(0, 50).map(toObservableNew).mergeAll();
+
+  function _next(x) { }
+  function _error(e) { }
+  function _complete() { }
+  return suite
+      .add('old map and mergeAll with immediate scheduler', function () {
+        _old.subscribe(_next, _error, _complete);
+      })
+      .add('new map and mergeAll with immediate scheduler', function () {
+        _new.subscribe(_next, _error, _complete);
+      });
+};

src/operator/map.ts

@@ -16,13 +16,33 @@ export function map<T, R>(project: (value: T, index: number) => R, thisArg?: any
   if (typeof project !== 'function') {
     throw new TypeError('argument is not a function. Are you looking for `mapTo()`?');
   }
-  return this.lift(new MapOperator(project, thisArg));
+
+  const source = this.source;
+  const operator = this.operator;
+
+  if (source && operator && operator.transduce) {
+    return source.lift(new MapOperator(operator.transduce(project, thisArg), thisArg));
+  } else {
+    return this.lift(new MapOperator(project, thisArg));
+  }
 }
 
 class MapOperator<T, R> implements Operator<T, R> {
   constructor(private project: (value: T, index: number) => R, private thisArg: any) {
   }
 
+  transduce(nextOperation: (value: T, index: number) => any, nextThisArg: any): (value: T, index: number) => R {
+    const fn = function transducedMap(value: T, index: number) {
+      const { project, thisArg, nextOperation, nextThisArg } = (<any>transducedMap);
+      return nextOperation.call(nextThisArg, project.call(thisArg, value, index), index);
+    };
+    (<any>fn).project = this.project;
+    (<any>fn).thisArg = this.thisArg;
+    (<any>fn).nextOperation = nextOperation;
+    (<any>fn).nextThisArg = nextThisArg;
+    return fn;
+  }
+
   call(subscriber: Subscriber<R>): Subscriber<T> {
     return new MapSubscriber(subscriber, this.project, this.thisArg);
   }

src/operator/mergeAll.ts

@@ -4,9 +4,15 @@ import {Observer} from '../Observer';
 import {Subscription} from '../Subscription';
 import {OuterSubscriber} from '../OuterSubscriber';
 import {subscribeToResult} from '../util/subscribeToResult';
+import {MergeMapOperator} from './mergeMap';
 
 export function mergeAll<T>(concurrent: number = Number.POSITIVE_INFINITY): T {
-  return this.lift(new MergeAllOperator(concurrent));
+  const { source, operator } = this;
+  if (source && operator && operator.transduce) {
+    return source.lift(new MergeMapOperator(operator.transduce((x: any) => x, this), null, concurrent));
+  } else {
+    return this.lift(new MergeAllOperator(concurrent));
+  }
 }
 
 export class MergeAllOperator<T> implements Operator<Observable<T>, T> {