repeat

Author: mattpodwysocki

src/modular/foo.js

@@ -0,0 +1,44 @@
+var $iterator$ = '@@iterator';
+
+function repeatValue(value, count) {
+  count == null && (count = -1);
+  return {
+    '@@iterator': function () {
+      return {
+        remaining: count,
+        next: function () {
+          if (this.remaining === 0) { return { done: true, value: undefined }; }
+          if (this.remaining > 0) { this.remaining--; }
+          return { done: false, value: value };
+        }
+      };
+    }
+  };
+}
+
+var e = repeatValue(42, 10)[$iterator$]();
+var result;
+do {
+  result = e.next();
+  console.log(result.value);
+} while (!result.done);
+
+function createDisposable(state) {
+  return {
+    isDisposed: false,
+    dispose: function () {
+      if (!this.isDisposed) {
+        this.isDisposed = true;
+        state.isDisposed = true;
+      }
+    }
+  };
+}
+
+var o = { isDisposed: false };
+var d = createDisposable(o);
+console.log(o.isDisposed);
+d.dispose();
+console.log(o.isDisposed);
+d.dispose();
+console.log(o.isDisposed);

src/modular/observable/repeat.js

@@ -0,0 +1,95 @@
+'use strict';
+
+var ObservableBase = require('./observablebase');
+var fromPromise = require('./frompromise');
+var isPromise = require('../helpers/ispromise');
+var AbstractObserver = require('../observer/abstractobserver');
+var NAryDisposable = require('../narydisposable');
+var SerialDisposable = require('../serialdisposable');
+var SingleAssignmentDisposable = require('../singleassignmentdisposable');
+var inherits = require('inherits');
+
+global._Rx || (global._Rx = {});
+if (!global._Rx.currentThreadScheduler) {
+  require('../scheduler/currentthreadscheduler');
+}
+
+var $iterator$ = '@@iterator';
+
+function repeatValue(value, count) {
+  count == null && (count = -1);
+  return {
+    '@@iterator': function () {
+      return {
+        remaining: count,
+        next: function () {
+          if (this.remaining === 0) { return { done: true, value: undefined }; }
+          if (this.remaining > 0) { this.remaining--; }
+          return { done: false, value: value };
+        }
+      };
+    }
+  };
+}
+
+function createDisposable(state) {
+  return {
+    isDisposed: false,
+    dispose: function () {
+      if (!this.isDisposed) {
+        this.isDisposed = true;
+        state.isDisposed = true;
+      }
+    }
+  };
+}
+
+function ConcatObserver(state, recurse) {
+  this._state = state;
+  this._recurse = recurse;
+  AbstractObserver.call(this);
+}
+
+inherits(ConcatObserver, AbstractObserver);
+
+ConcatObserver.prototype.next = function (x) { this._state.o.onNext(x); };
+ConcatObserver.prototype.error = function (e) { this._state.o.onError(e); };
+ConcatObserver.prototype.completed = function () { this._recurse(this._state); };
+
+function ConcatObservable(sources) {
+  this.sources = sources;
+  ObservableBase.call(this);
+}
+
+inherits(ConcatObservable, ObservableBase);
+
+function scheduleMethod(state, recurse) {
+  if (state.isDisposed) { return; }
+  var currentItem = state.e.next();
+  if (currentItem.done) { return state.o.onCompleted(); }
+
+  // Check if promise
+  var currentValue = currentItem.value;
+  isPromise(currentValue) && (currentValue = fromPromise(currentValue));
+
+  var d = new SingleAssignmentDisposable();
+  state.subscription.setDisposable(d);
+  d.setDisposable(currentValue.subscribe(new ConcatObserver(state, recurse)));
+}
+
+ConcatObservable.prototype.subscribeCore = function (o) {
+  var subscription = new SerialDisposable();
+  var state = {
+    isDisposed: false,
+    o: o,
+    subscription: subscription,
+    e: this.sources[$iterator$]()
+  };
+
+  var cancelable = global._Rx.currentThreadScheduler.scheduleRecursive(state, scheduleMethod);
+  return new NAryDisposable([subscription, cancelable, createDisposable(state)]);
+};
+
+module.exports = function repeat(source, repeatCount) {
+  return new ConcatObservable(repeatValue(source, repeatCount));
+};

src/modular/observable/repeatvalue.js

@@ -0,0 +1,15 @@
+'use strict';
+
+var just = require('./just');
+var repeat = require('./repeat');
+var isScheduler = require('../scheduler').isScheduler;
+
+global._Rx || (global._Rx = {});
+if (!global._Rx.currentThreadScheduler) {
+  require('../scheduler/currentthreadscheduler');
+}
+
+module.exports = function repeatValue(value, repeatCount, scheduler) {
+  isScheduler(scheduler) || (scheduler = global._Rx.currentThreadScheduler);
+  return repeat(just(value, scheduler), repeatCount);
+};

src/modular/observable/replay.js

@@ -0,0 +1,11 @@
+'use strict';
+
+var multicast = require('./multicast');
+var ReplaySubject = require('../replaysubject');
+var isFunction = require('../helpers/isfunction');
+
+module.exports = function replay (source, selector, bufferSize, windowSize, scheduler) {
+  return isFunction(selector) ?
+    multicast(source, function () { return new ReplaySubject(bufferSize, windowSize, scheduler); }, selector) :
+    multicast(source, new ReplaySubject(bufferSize, windowSize, scheduler));
+};

src/modular/observable/retry.js

@@ -7,7 +7,6 @@ var AbstractObserver = require('../observer/abstractobserver');
 var NAryDisposable = require('../narydisposable');
 var SerialDisposable = require('../serialdisposable');
 var SingleAssignmentDisposable = require('../singleassignmentdisposable');
-var tryCatch = require('../internal/trycatchutils').tryCatch;
 var inherits = require('inherits');
 
 global._Rx || (global._Rx = {});
@@ -18,6 +17,7 @@ if (!global._Rx.currentThreadScheduler) {
 var $iterator$ = '@@iterator';
 
 function repeat(value, count) {
+  count == null && (count = -1);
   return {
     '@@iterator': function () {
       return {
@@ -65,8 +65,7 @@ inherits(CatchErrorObservable, ObservableBase);
 
 function scheduleMethod(state, recurse) {
   if (state.isDisposed) { return; }
-  var currentItem = tryCatch(state.e.next).call(state.e);
-  if (currentItem === global._Rx.errorObj) { return state.o.onError(currentItem.e); }
+  var currentItem = state.e.next();
   if (currentItem.done) { return state.lastError !== null ? state.o.onError(state.lastError) : state.o.onCompleted(); }
 
   var currentValue = currentItem.value;