takeLast

Author: mattpodwysocki

src/modular/index.js

@@ -135,6 +135,7 @@ Observable.addToPrototype({
   switch: require('./observable/switch'),
   switchFirst: require('./observable/switchfirst'),
   take: require('./observable/take'),
+  takeLast: require('./observable/takelast'),
   takeLastBuffer: require('./observable/takelastbuffer'),
   takeLastBufferTime: require('./observable/takelastbufferwithtime'),
   lastLastTime: require('./observable/takelastwithtime'),

src/modular/observable/range.js

@@ -13,6 +13,19 @@ 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) {
@@ -25,6 +38,13 @@ 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/takelast.js

@@ -0,0 +1,83 @@
+'use strict';
+
+var ObservableBase = require('./observablebase');
+var AbstractObserver = require('../observer/abstractobserver');
+var Scheduler = require('../scheduler');
+var BinaryDisposable = require('../binarydisposable');
+var SingleAssignmentDisposable = require('../singleassignmentdisposable');
+var ArgumentOutOfRangeError = require('../internal/errors').ArgumentOutOfRangeError;
+var inherits = require('inherits');
+
+function TakeLastObserver(o, c, s, ss, ls) {
+  this._o = o;
+  this._c = c;
+  this._s = s;
+  this._ls = ls;
+  this._ss = ss;
+  this._q = [];
+  AbstractObserver.call(this);
+}
+
+inherits(TakeLastObserver, AbstractObserver);
+
+TakeLastObserver.prototype.next = function (x) {
+  this._q.push(x);
+  this._q.length > this._c && this._q.shift();
+};
+
+TakeLastObserver.prototype.error = function (e) {
+  this._o.onError(e);
+};
+
+function loopRecursive(state, recurse) {
+  if (state[1].length > 0) {
+    state[0].onNext(state[1].shift());
+    recurse(state);
+  } else {
+    state[0].onCompleted();
+  }
+}
+
+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));
+  }
+};
+
+function TakeLastObservable(source, count, scheduler) {
+  this.source = source;
+  this.count = count;
+  this.scheduler = scheduler;
+  ObservableBase.call(this);
+}
+
+inherits(TakeLastObservable, ObservableBase);
+
+TakeLastObservable.prototype.subscribeCore = function (o) {
+  var subscription = new SingleAssignmentDisposable();
+  var loopSubscription = new SingleAssignmentDisposable();
+  subscription.setDisposable(this.source.subscribe(new TakeLastObserver(o, this.count, this.scheduler, subscription, loopSubscription)));
+
+  return new BinaryDisposable(subscription, loopSubscription);
+};
+
+module.exports = function takeLast (source, count, scheduler) {
+  if (count < 0) { throw new ArgumentOutOfRangeError(); }
+  Scheduler.isScheduler(scheduler) || (scheduler = Scheduler.queue);
+  return new TakeLastObservable(source, count, scheduler);
+};

src/modular/scheduler/defaultscheduler.js

@@ -175,8 +175,8 @@ function scheduleLongRunning(state, action, disposable) {
 
 DefaultScheduler.prototype.scheduleLongRunning = function (state, action) {
   var disposable = Disposable.create(noop);
-  scheduleMethod(scheduleLongRunning(state, action, disposable));
-  return disposable;
+  var id = scheduleMethod(scheduleLongRunning(state, action, disposable));
+  return new BinaryDisposable(disposable, new ClearDisposable(clearMethod, id));
 };
 
 module.exports = DefaultScheduler;