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);
  }
 }