windowed/stopAndWait

Author: mattpodwysocki

src/modular/observable/stopandwait.js

@@ -5,7 +5,7 @@ var AbstractObserver = require('../observer/abstractobserver');
 var BinaryDisposable = require('../binarydisposable');
 var inherits = require('inherits');
 
-function StopAndWaitObserver (observer, observable, scheduler, cancel) {
+function StopAndWaitObserver(observer, observable, scheduler, cancel) {
   this.observer = observer;
   this.observable = observable;
   this.scheduler = scheduler;
@@ -14,6 +14,8 @@ function StopAndWaitObserver (observer, observable, scheduler, cancel) {
   AbstractObserver.call(this);
 }
 
+inherits(StopAndWaitObserver, AbstractObserver);
+
 StopAndWaitObserver.prototype.completed = function () {
   this.observer.onCompleted();
   this.dispose();
@@ -33,7 +35,7 @@ StopAndWaitObserver.prototype.next = function (value) {
   this.scheduleDisposable = this.scheduler.schedule(this, innerScheduleMethod);
 };
 
-StopAndWaitObserver.dispose = function () {
+StopAndWaitObserver.prototype.dispose = function () {
   this.observer = null;
   if (this.cancel) {
     this.cancel.dispose();
@@ -46,7 +48,7 @@ StopAndWaitObserver.dispose = function () {
   AbstractObserver.prototype.dispose.call(this);
 };
 
-function StopAndWaitObservable (source, scheduler) {
+function StopAndWaitObservable(source, scheduler) {
   this.source = source;
   this.scheduler = scheduler;
   Observable.call(this);

src/modular/test/stopandwait.js

@@ -0,0 +1,182 @@
+'use strict';
+
+var test = require('tape');
+var Observable = require('../observable');
+var TestScheduler = require('../testing/testscheduler');
+var reactiveAssert = require('../testing/reactiveassert');
+var ReactiveTest = require('../testing/reactivetest');
+var onNext = ReactiveTest.onNext,
+  onError = ReactiveTest.onError,
+  onCompleted = ReactiveTest.onCompleted,
+  subscribe = ReactiveTest.subscribe;
+
+Observable.addToPrototype({
+  controlled: require('../observable/controlled')
+});
+
+test('ControlledObservable#stopAndWait never', function (t) {
+  var scheduler = new TestScheduler();
+
+  var xs = scheduler.createHotObservable(
+    onNext(150, 1)
+  );
+
+  var results = scheduler.startScheduler(function () {
+    return xs.controlled(true, scheduler).stopAndWait(scheduler);
+  });
+
+  reactiveAssert(t, results.messages, [
+
+  ]);
+
+  reactiveAssert(t, xs.subscriptions, [
+    subscribe(200, 1000)
+  ]);
+
+  t.end();
+});
+
+test('ControlledObservable#stopAndWait empty', function (t) {
+  var scheduler = new TestScheduler();
+
+  var xs = scheduler.createHotObservable(
+    onNext(150, 1),
+    onCompleted(250)
+  );
+
+  var results = scheduler.startScheduler(function () {
+    return xs.controlled(true, scheduler).stopAndWait(scheduler);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onCompleted(250)
+  ]);
+
+  reactiveAssert(t, xs.subscriptions, [
+    subscribe(200, 250)
+  ]);
+
+  t.end();
+});
+
+test('ControlledObservable#stopAndWait basic', function (t) {
+  var scheduler = new TestScheduler();
+
+  var xs = scheduler.createHotObservable(
+    onNext(150, 1),
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4),
+    onNext(240, 5),
+    onCompleted(250)
+  );
+
+  var results = scheduler.startScheduler(function () {
+    return xs.controlled(true, scheduler).stopAndWait(scheduler);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4),
+    onNext(240, 5),
+    onCompleted(250)
+  ]);
+
+  reactiveAssert(t, xs.subscriptions, [
+    subscribe(200, 250)
+  ]);
+
+  t.end();
+});
+
+test('ControlledObservable#stopAndWait error', function (t) {
+  var error = new Error();
+
+  var scheduler = new TestScheduler();
+
+  var xs = scheduler.createHotObservable(
+    onNext(150, 1),
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4),
+    onNext(240, 5),
+    onError(250, error)
+  );
+
+  var results = scheduler.startScheduler(function () {
+    return xs.controlled(true, scheduler).stopAndWait(scheduler);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4),
+    onNext(240, 5),
+    onError(250, error)
+  ]);
+
+  reactiveAssert(t, xs.subscriptions, [
+    subscribe(200, 250)
+  ]);
+
+  t.end();
+});
+
+test('ControlledObservable#stopAndWait infinite', function (t) {
+  var scheduler = new TestScheduler();
+
+  var xs = scheduler.createHotObservable(
+    onNext(150, 1),
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4),
+    onNext(240, 5)
+  );
+
+  var results = scheduler.startScheduler(function () {
+    return xs.controlled(true, scheduler).stopAndWait(scheduler);
+  });
+
+  reactiveAssert(t, results.messages, [
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4),
+    onNext(240, 5)
+  ]);
+
+  reactiveAssert(t, xs.subscriptions, [
+    subscribe(200, 1000)
+  ]);
+
+  t.end();
+});
+
+test('ControlledObservable#windowed disposed', function (t) {
+  var scheduler = new TestScheduler();
+
+  var xs = scheduler.createHotObservable(
+    onNext(150, 1),
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4),
+    onNext(240, 5),
+    onNext(250)
+  );
+
+  var results = scheduler.startScheduler(function () {
+    return xs.controlled(true, scheduler).stopAndWait(scheduler);
+  }, { disposed: 235 });
+
+  reactiveAssert(t, results.messages, [
+    onNext(210, 2),
+    onNext(220, 3),
+    onNext(230, 4)
+  ]);
+
+  reactiveAssert(t, xs.subscriptions, [
+    subscribe(200, 235)
+  ]);
+
+  t.end();
+});