Removing longrunning

Author: mattpodwysocki

src/modular/observable/from.js

@@ -100,7 +100,7 @@ function FromObservable(iterable, fn, scheduler) {
 
 inherits(FromObservable, ObservableBase);
 
-function createScheduleMethod(o, it, fn) {
+function scheduleRecursive(o, it, fn) {
   return function loopRecursive(i, recurse) {
     var next = tryCatch(it.next).call(it);
     if (next === errorObj) { return o.onError(next.e); }
@@ -122,7 +122,7 @@ FromObservable.prototype.subscribeCore = function (o) {
   var list = Object(this._iterable),
       it = getIterable(list);
 
-  return this._scheduler.scheduleRecursive(0, createScheduleMethod(o, it, this._fn));
+  return this._scheduler.scheduleRecursive(0, scheduleRecursive(o, it, this._fn));
 };
 
 /**

src/modular/observable/fromarray.js

@@ -4,15 +4,13 @@ var ObservableBase = require('./observablebase');
 var Scheduler = require('../scheduler');
 var inherits = require('inherits');
 
-function scheduleMethod(o, args) {
-  return function loopRecursive (i, recurse) {
-    if (i < args.length) {
-      o.onNext(args[i]);
-      recurse(i + 1);
-    } else {
-      o.onCompleted();
-    }
-  };
+function scheduleRecursive(state, recurse) {
+  if (state.i < state.len) {
+    state.o.onNext(state.args[state.i++]);
+    recurse(state);
+  } else {
+    state.o.onCompleted();
+  }
 }
 
 function FromArrayObservable(args, scheduler) {
@@ -24,7 +22,13 @@ function FromArrayObservable(args, scheduler) {
 inherits(FromArrayObservable, ObservableBase);
 
 FromArrayObservable.prototype.subscribeCore = function (o) {
-  return this._scheduler.scheduleRecursive(0, scheduleMethod(o, this._args));
+  var state = {
+    i: 0,
+    args: this._args,
+    len: this._args.length,
+    o: o
+  };
+  return this._scheduler.scheduleRecursive(state, scheduleRecursive);
 };
 
 module.exports = function fromArray(array, scheduler) {

src/modular/observable/range.js

@@ -13,19 +13,6 @@ function RangeObservable(start, count, scheduler) {
 
 inherits(RangeObservable, ObservableBase);
 
-function loopLongRunning(start, count, o) {
-  return function loop (i, cancel) {
-    while (!cancel.isDisposed && i < count) {
-      o.onNext(start + i);
-      i++;
-    }
-
-    if (!cancel.isDisposed) {
-      o.onCompleted();
-    }
-  };
-}
-
 function loopRecursive(start, count, o) {
   return function loop (i, recurse) {
     if (i < count) {
@@ -38,13 +25,6 @@ function loopRecursive(start, count, o) {
 }
 
 RangeObservable.prototype.subscribeCore = function (o) {
-  if (this.scheduler.scheduleLongRunning) {
-    return this.scheduler.scheduleLongRunning(
-      0,
-      loopLongRunning(this.start, this.rangeCount, o)
-    );
-  }
-
   return this.scheduler.scheduleRecursive(
     0,
     loopRecursive(this.start, this.rangeCount, o)

src/modular/observable/repeatvalue.js

@@ -1,10 +1,35 @@
 'use strict';
 
-var just = require('./just');
-var repeat = require('./repeat');
+var ObservableBase = require('./observablebase');
 var Scheduler = require('../scheduler');
+var inherits = require('inherits');
 
-module.exports = function repeatValue(value, repeatCount, scheduler) {
+function RepeatValueObservable(value, count, scheduler) {
+  this._value = value;
+  this._count = count;
+  this._scheduler = scheduler;
+  ObservableBase.call(this);
+}
+
+inherits(RepeatValueObservable, ObservableBase);
+
+function scheduleRecursive(state, recurse) {
+  if (state.n === 0) { return state.o.onCompleted(); }
+  if (state.n > 0) { state.n--; }
+  state.o.onNext(state.value);
+  recurse(state);
+}
+
+RepeatValueObservable.prototype.subscribeCore = function (o) {
+  var state = {
+    value: this._value,
+    n: this._count,
+    o: o
+  };
+  return this._scheduler.scheduleRecursive(state, scheduleRecursive);
+};
+
+module.exports = function repeatValue (value, repeatCount, scheduler) {
   Scheduler.isScheduler(scheduler) || (scheduler = Scheduler.queue);
-  return repeat(just(value, scheduler), repeatCount);
+  return new RepeatValueObservable(value, repeatCount, scheduler);
 };

src/modular/observable/takelast.js

@@ -38,25 +38,9 @@ function loopRecursive(state, recurse) {
   }
 }
 
-function loopLongRunning(state, cancel) {
-  var o = state[0], q = state[1], n = q.length;
-  while(!cancel.isDisposed) {
-    if (n === 0) {
-      o.onCompleted();
-    } else {
-      o.onNext(q.shift());
-    }
-    n--;
-  }
-}
-
 TakeLastObserver.prototype.completed = function () {
   this._ss.dispose();
-  if (this._s.scheduleLongRunning) {
-    this._ls.setDisposable(this._s.scheduleLongRunning([this._o, this._q], loopLongRunning));
-  } else {
-    this._ls.setDisposable(this._s.scheduleRecursive([this._o, this._q], loopRecursive));
-  }
+  this._ls.setDisposable(this._s.scheduleRecursive([this._o, this._q], loopRecursive));
 };
 
 function TakeLastObservable(source, count, scheduler) {

src/modular/scheduler/defaultscheduler.js

@@ -169,14 +169,4 @@ DefaultScheduler.prototype._scheduleFuture = function (state, dueTime, action) {
   return new BinaryDisposable(disposable, new ClearDisposable(global.clearTimeout, id));
 };
 
-function scheduleLongRunning(state, action, disposable) {
-  return function () { action(state, disposable); };
-}
-
-DefaultScheduler.prototype.scheduleLongRunning = function (state, action) {
-  var disposable = Disposable.create(noop);
-  var id = scheduleMethod(scheduleLongRunning(state, action, disposable));
-  return new BinaryDisposable(disposable, new ClearDisposable(clearMethod, id));
-};
-
 module.exports = DefaultScheduler;

src/modular/test/repeat.js

@@ -29,7 +29,7 @@ test('Observable#repeat value count zero', function (t) {
   });
 
   reactiveAssert(t, results.messages, [
-    onCompleted(200)
+    onCompleted(201)
   ]);
 
   t.end();
@@ -44,7 +44,7 @@ test('Observable#repeat value count one', function (t) {
 
   reactiveAssert(t, results.messages, [
     onNext(201, 42),
-    onCompleted(201)
+    onCompleted(202)
   ]);
 
   t.end();
@@ -68,7 +68,7 @@ test('Observable#repeat value count ten', function (t) {
     onNext(208, 42),
     onNext(209, 42),
     onNext(210, 42),
-    onCompleted(210)
+    onCompleted(211)
   ]);
 
   t.end();
@@ -121,7 +121,8 @@ test('Observable#repeat Observable basic', function (t) {
     onNext(100, 1),
     onNext(150, 2),
     onNext(200, 3),
-    onCompleted(250));
+    onCompleted(250)
+  );
 
   var results = scheduler.startScheduler(function () {
     return xs.repeat();

tests/perf/operators/fromarray.js

@@ -1,5 +1,5 @@
 var RxOld = require('../old/rx.lite');
-var RxNew = require('../../../dist/rx.lite');
+var RxNew = require('../../../src/modular/.');
 var Benchmark = require('benchmark');
 
 var suite = new Benchmark.Suite;

tests/perf/operators/range.js

@@ -1,5 +1,5 @@
-var RxOld = require('../old/rx.lite');
-var RxNew = require('../../../dist/rx.lite');
+var RxOld = require('../../../dist/rx.lite');
+var RxNew = require('../../../src/modular/.');
 var Benchmark = require('benchmark');
 
 var suite = new Benchmark.Suite;

tests/perf/operators/repeat.js

@@ -1,25 +1,9 @@
-var RxOld = require('../old/rx.lite');
-var RxNew = require('../../../dist/rx.lite');
+var RxOld = require('../../../dist/rx.lite');
+var RxNew = require('../../../src/modular/.');
 var Benchmark = require('benchmark');
 
 var suite = new Benchmark.Suite;
 
-// Backfill range to get rid of differences
-RxOld.range = function (start, count) {
-  var scheduler = RxNew.Scheduler.currentThread;
-  return new RxNew.AnonymousObservable(function (observer) {
-    return scheduler.scheduleRecursive(0, function (i, self) {
-      if (i < count) {
-        observer.onNext(start + i);
-        self(i + 1);
-      } else {
-        observer.onCompleted();
-      }
-    });
-  });
-};
-RxNew.range = RxOld.range;
-
 // add tests
 suite.add('old', function() {
   RxOld.Observable.repeat(42, 25).subscribe();