fix(groupBy): does not emit on unsubscribed group

ReactiveX · Feb 9, 2016 · 6d08705 · 6d08705
1 parent 29b630b
commit 6d08705
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 9 deletions.
diff --git a/spec/operators/groupBy-spec.js b/spec/operators/groupBy-spec.js
@@ -1,4 +1,4 @@
-/* globals describe, it, expect, hot, cold, expectObservable, expectSubscriptions */
+/* globals describe, it, expect, hot, cold, expectObservable, expectSubscriptions, rxTestScheduler */
 var Rx = require('../../dist/cjs/Rx.KitchenSink');
 var Observable = Rx.Observable;
 var GroupedObservable = require('../../dist/cjs/operator/groupBy').GroupedObservable;

diff --git a/src/operator/groupBy.ts b/src/operator/groupBy.ts
@@ -83,7 +83,7 @@ class GroupBySubscriber<T, K, R> extends Subscriber<T> implements RefCountSubscr
 
     if (!group) {
       groups.set(key, group = new Subject<R>());
-      let groupedObservable = new GroupedObservable(key, group, this);
+      const groupedObservable = new GroupedObservable(key, group, this);
 
       if (this.durationSelector) {
         this._selectDuration(key, group);
@@ -95,7 +95,7 @@ class GroupBySubscriber<T, K, R> extends Subscriber<T> implements RefCountSubscr
     if (this.elementSelector) {
       this._selectElement(value, group);
     } else {
-      group.next(value);
+      this.tryGroupNext(value, group);
     }
   }
 
@@ -107,7 +107,7 @@ class GroupBySubscriber<T, K, R> extends Subscriber<T> implements RefCountSubscr
       this.error(err);
       return;
     }
-    group.next(result);
+    this.tryGroupNext(result, group);
   }
 
   private _selectDuration(key: K, group: any) {
@@ -121,6 +121,12 @@ class GroupBySubscriber<T, K, R> extends Subscriber<T> implements RefCountSubscr
     this.add(duration.subscribe(new GroupDurationSubscriber(key, group, this)));
   }
 
+  private tryGroupNext(value: T|R, group: Subject<T | R>): void {
+    if (!group.isUnsubscribed) {
+      group.next(value);
+    }
+  }
+
   protected _error(err: any): void {
     const groups = this.groups;
     if (groups) {
@@ -165,17 +171,30 @@ class GroupDurationSubscriber<K, T> extends Subscriber<T> {
   }
 
   protected _next(value: T): void {
-    this.group.complete();
-    this.parent.removeGroup(this.key);
+    this.tryComplete();
   }
 
   protected _error(err: any): void {
-    this.group.error(err);
-    this.parent.removeGroup(this.key);
+    this.tryError(err);
   }
 
   protected _complete(): void {
-    this.group.complete();
+    this.tryComplete();
+  }
+
+  private tryError(err: any): void {
+    const group = this.group;
+    if (!group.isUnsubscribed) {
+      group.error(err);
+    }
+    this.parent.removeGroup(this.key);
+  }
+
+  private tryComplete(): void {
+    const group = this.group;
+    if (!group.isUnsubscribed) {
+      group.complete();
+    }
     this.parent.removeGroup(this.key);
   }
 }