fix(SubjectSource): allow source to be run after being disposed

A source may be run and disposed multiple times in its lifetime, but BasicSubjectSource refuses to
relay events after it is run, disposed, and run again. This is a fix for that issue.

A most stream may be observed anew after ended by completion or error, so this update allows the
same. In addition, a subject created by holdSubject() does not begin buffering events until it has
an observer.

Author: brandonpayton

src/SubjectSource.ts

@@ -14,11 +14,14 @@ export interface SubjectSource<T> extends Source<T> {
 export class BasicSubjectSource<T> implements SubjectSource<T> {
   protected scheduler: Scheduler = defaultScheduler;
   protected sinks: Sink<T>[] = [];
-  protected active: boolean = true;
+  protected active: boolean = false;
 
   run (sink: Sink<T>, scheduler: Scheduler): Disposable<T> {
     const n = this.add(sink);
-    if (n === 1) this.scheduler = scheduler;
+    if (n === 1) {
+      this.scheduler = scheduler;
+      this.active = true;
+    }
     return new SubjectDisposable<T>(this, sink);
   }
 
@@ -48,14 +51,14 @@ export class BasicSubjectSource<T> implements SubjectSource<T> {
   error (err: Error): void {
     if (!this.active || this.scheduler === void 0) return;
 
-    this.active = false;
+    this._dispose();
     this._error(this.scheduler.now(), err);
   }
 
   complete (value?: T): void {
     if (!this.active || this.scheduler === void 0) return;
 
-    this.active = false;
+    this._dispose();
     this._complete(this.scheduler.now(), value);
   }
 

test/holdSubject.js

@@ -25,6 +25,9 @@ describe('holdSubject', () => {
   it('should allow for adjusting bufferSize of stream', () => {
     const stream = holdSubject(3)
 
+    // Add an observer so the stream begins buffering events
+    stream.observe(() => {})
+
     stream.next(1)
     stream.next(2)
     stream.next(3)

test/subject.js

@@ -75,7 +75,7 @@ describe('subject()', () => {
       stream.complete()
     })
 
-    it('should not allow events after end', done => {
+    it('should not notify existing observers after end', done => {
       const stream = subject()
 
       stream
@@ -87,6 +87,53 @@ describe('subject()', () => {
       stream.next(1)
     })
 
+    it('should not notify existing observers after error', done => {
+      const stream = subject()
+
+      stream
+        .forEach(assert.fail)
+        .then(done, () => done())
+        .catch(assert.fail)
+
+      stream.error(new Error())
+      stream.next(1)
+    })
+
+    it('should allow new observers after end', done => {
+      const stream = subject()
+
+      stream.complete()
+
+      stream
+        .reduce((x, y) => x.concat(y), [])
+        .then(x => assert.deepEqual(x, [1, 2, 3]))
+        .then(done)
+
+      stream.next(1)
+      stream.next(2)
+      stream.next(3)
+
+      stream.complete()
+    })
+
+
+    it('should allow new observers after error', done => {
+      const stream = subject()
+
+      stream.error(new Error())
+
+      stream
+        .reduce((x, y) => x.concat(y), [])
+        .then(x => assert.deepEqual(x, [1, 2, 3]))
+        .then(done)
+
+      stream.next(1)
+      stream.next(2)
+      stream.next(3)
+
+      stream.complete()
+    })
+
     it('should support transient use as a signal stream', done => {
       const stream = subject()
       const signalStream = subject()