fix(groupBy): fix groupBy to use lift(), supports composability

Fix bug #1085 in which groupBy was not using lift(). Using lift() is critical to support the ubiquitous lift-based architecture in RxJS Next Resolves #1085.
ReactiveX · Jan 27, 2016 · 815cfae · 815cfae
1 parent 9eb25ba
commit 815cfae
Show file tree

Hide file tree

Showing 4 changed files with 95 additions and 120 deletions.
diff --git a/spec/Observable-spec.js b/spec/Observable-spec.js
@@ -331,17 +331,16 @@ describe('Observable.lift', function () {
 
  it('should allow injecting behaviors into all subscribers in an operator ' +
  'chain when overriden', function (done) {
- // The custom Observable
- function LogObservable() {
- Observable.apply(this, arguments);
+ // The custom Subscriber
+ var log = [];
+ function LogSubscriber() {
+ Subscriber.apply(this, arguments);
  }
- LogObservable.prototype = Object.create(Observable.prototype);
- LogObservable.prototype.constructor = LogObservable;
- LogObservable.prototype.lift = function (operator) {
- var obs = new LogObservable();
- obs.source = this;
- obs.operator = new LogOperator(operator);
- return obs;
+ LogSubscriber.prototype = Object.create(Subscriber.prototype);
+ LogSubscriber.prototype.constructor = LogSubscriber;
+ LogSubscriber.prototype.next = function (x) {
+ log.push('next ' + x);
+ this.destination.next(x);
  };
 
  // The custom Operator
@@ -352,16 +351,17 @@ describe('Observable.lift', function () {
  return this.childOperator.call(new LogSubscriber(subscriber));
  };
 
- // The custom Subscriber
- var log = [];
- function LogSubscriber() {
- Subscriber.apply(this, arguments);
+ // The custom Observable
+ function LogObservable() {
+ Observable.apply(this, arguments);
  }
- LogSubscriber.prototype = Object.create(Subscriber.prototype);
- LogSubscriber.prototype.constructor = LogSubscriber;
- LogSubscriber.prototype.next = function (x) {
- log.push('next ' + x);
- this.destination.next(x);
+ LogObservable.prototype = Object.create(Observable.prototype);
+ LogObservable.prototype.constructor = LogObservable;
+ LogObservable.prototype.lift = function (operator) {
+ var obs = new LogObservable();
+ obs.source = this;
+ obs.operator = new LogOperator(operator);
+ return obs;
  };
 
  // Use the LogObservable
@@ -372,7 +372,7 @@ describe('Observable.lift', function () {
  observer.complete();
  })
  .map(function (x) { return 10 * x; })
- .filter(function (x) { return x > 15 })
+ .filter(function (x) { return x > 15; })
  .count();
 
  expect(result instanceof LogObservable).toBe(true);

diff --git a/spec/operators/groupBy-spec.js b/spec/operators/groupBy-spec.js
@@ -1387,4 +1387,4 @@ describe('Observable.prototype.groupBy()', function () {
  });
  }, done.fail, done);
  });
-});
+});
diff --git a/src/Observable.ts b/src/Observable.ts
@@ -215,7 +215,7 @@ export class Observable<T> implements CoreOperators<T> {
  projectResult?: (x: T, y: any, ix: number, iy: number) => R,
  concurrent?: number) => Observable<R>;
  flatMapTo: <R>(observable: Observable<any>, projectResult?: (x: T, y: any, ix: number, iy: number) => R, concurrent?: number) => Observable<R>;
- groupBy: <K, R>(keySelector: (value: T) => string,
+ groupBy: <K, R>(keySelector: (value: T) => K,
  elementSelector?: (value: T) => R,
  durationSelector?: (group: GroupedObservable<K, R>) => Observable<any>) => Observable<GroupedObservable<K, R>>;
  ignoreElements: () => Observable<T>;

diff --git a/src/operator/groupBy.ts b/src/operator/groupBy.ts
@@ -1,9 +1,12 @@
 import {Subscriber} from '../Subscriber';
 import {Subscription} from '../Subscription';
 import {Observable} from '../Observable';
+import {Operator} from '../Operator';
 import {Subject} from '../Subject';
 import {Map} from '../util/Map';
 import {FastMap} from '../util/FastMap';
+import {tryCatch} from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
 
 /**
  * Groups the items emitted by an Observable according to a specified criterion,
@@ -19,33 +22,38 @@ import {FastMap} from '../util/FastMap';
  */
 export function groupBy<T, K, R>(keySelector: (value: T) => K,
  elementSelector?: (value: T) => R,
- durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>): GroupByObservable<T, K, R> {
- return new GroupByObservable(this, keySelector, elementSelector, durationSelector);
+ durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>): Observable<GroupedObservable<K, R>> {
+ return this.lift(new GroupByOperator(this, keySelector, elementSelector, durationSelector));
 }
 
-export class GroupByObservable<T, K, R> extends Observable<GroupedObservable<K, R>> {
+export interface RefCountSubscription {
+ count: number;
+ unsubscribe: () => void;
+ isUnsubscribed: boolean;
+ attemptedToUnsubscribe: boolean;
+}
+
+class GroupByOperator<T, K, R> extends Operator<T, R> {
  constructor(public source: Observable<T>,
  private keySelector: (value: T) => K,
  private elementSelector?: (value: T) => R,
  private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>) {
  super();
  }
 
- protected _subscribe(subscriber: Subscriber<any>): Subscription {
- const refCountSubscription = new RefCountSubscription();
- const groupBySubscriber = new GroupBySubscriber(
- subscriber, refCountSubscription, this.keySelector, this.elementSelector, this.durationSelector
+ call(subscriber: Subscriber<GroupedObservable<K, R>>): Subscriber<T> {
+ return new GroupBySubscriber(
+ subscriber, this.keySelector, this.elementSelector, this.durationSelector
  );
- refCountSubscription.setPrimary(this.source.subscribe(groupBySubscriber));
- return refCountSubscription;
  }
 }
 
-class GroupBySubscriber<T, K, R> extends Subscriber<T> {
+class GroupBySubscriber<T, K, R> extends Subscriber<T> implements RefCountSubscription {
  private groups: Map<K, Subject<T|R>> = null;
+ public attemptedToUnsubscribe: boolean = false;
+ public count: number = 0;
 
- constructor(destination: Subscriber<R>,
- private refCountSubscription: RefCountSubscription,
+ constructor(destination: Subscriber<GroupedObservable<K, R>>,
  private keySelector: (value: T) => K,
  private elementSelector?: (value: T) => R,
  private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>) {
@@ -54,67 +62,48 @@ class GroupBySubscriber<T, K, R> extends Subscriber<T> {
  this.add(destination);
  }
 
- protected _next(value: T): void {
- let key: any;
- try {
- key = this.keySelector(value);
- } catch (err) {
- this.error(err);
- return;
- }
- this._group(value, key);
- }
-
- private _group(value: T, key: K) {
- let groups = this.groups;
+ protected _next(x: T): void {
+ let key = tryCatch(this.keySelector)(x);
+ if (key === errorObject) {
+ this.error(errorObject.e);
+ } else {
+ let groups = this.groups;
+ const elementSelector = this.elementSelector;
+ const durationSelector = this.durationSelector;
 
- if (!groups) {
- groups = this.groups = typeof key === 'string' ? new FastMap() : new Map();
- }
+  if (!groups) {
+  groups = this.groups = typeof key === 'string' ? new FastMap() : new Map();
+  }
 
- let group = groups.get(key);
+  let group = groups.get(key);
 
- if (!group) {
- groups.set(key, group = new Subject<T|R>());
- let groupedObservable = new GroupedObservable(key, group, this.refCountSubscription);
+  if (!group) {
+  groups.set(key, group = new Subject<R>());
+  let groupedObservable = new GroupedObservable(key, group, this);
 
- if (this.durationSelector) {
- if (!this._tryDuration(key, group)) {
- return;
+ if (durationSelector) {
+ let duration = tryCatch(durationSelector)(new GroupedObservable<K, R>(key, <any>group));
+ if (duration === errorObject) {
+ this.error(errorObject.e);
+ } else {
+ this.add(duration.subscribe(new GroupDurationSubscriber(key, group, this)));
+ }
  }
- }
 
- this.destination.next(groupedObservable);
- }
-
- if (this.elementSelector) {
- this._tryElementSelector(value, group);
- } else {
- group.next(value);
- }
- }
-
- private _tryElementSelector(value: T, group: Subject<T | R>) {
- let result: any;
- try {
- result = this.elementSelector(value);
- } catch (err) {
- this.error(err);
- return;
- }
- group.next(result);
- }
+ this.destination.next(groupedObservable);
+ }
 
- private _tryDuration(key: K, group: any): boolean {
- let duration: any;
- try {
- duration = this.durationSelector(new GroupedObservable<K, R>(key, group));
- } catch (err) {
- this.error(err);
- return false;
+ if (elementSelector) {
+ let value = tryCatch(elementSelector)(x);
+ if (value === errorObject) {
+ this.error(errorObject.e);
+ } else {
+ group.next(value);
+ }
+ } else {
+ group.next(x);
+ }
  }
- this.add(duration.subscribe(new GroupDurationSubscriber(key, group, this)));
- return true;
  }
 
  protected _error(err: any): void {
@@ -142,6 +131,15 @@ class GroupBySubscriber<T, K, R> extends Subscriber<T> {
  removeGroup(key: K): void {
  this.groups.delete(key);
  }
+
+ unsubscribe() {
+ if (!this.isUnsubscribed && !this.attemptedToUnsubscribe) {
+ this.attemptedToUnsubscribe = true;
+ if (this.count === 0) {
+ super.unsubscribe();
+ }
+ }
+ }
 }
 
 class GroupDurationSubscriber<K, T> extends Subscriber<T> {
@@ -167,30 +165,6 @@ class GroupDurationSubscriber<K, T> extends Subscriber<T> {
  }
 }
 
-export class RefCountSubscription extends Subscription {
- primary: Subscription;
- attemptedToUnsubscribePrimary: boolean = false;
- count: number = 0;
-
- constructor() {
- super();
- }
-
- setPrimary(subscription: Subscription) {
- this.primary = subscription;
- }
-
- unsubscribe() {
- if (!this.isUnsubscribed && !this.attemptedToUnsubscribePrimary) {
- this.attemptedToUnsubscribePrimary = true;
- if (this.count === 0) {
- super.unsubscribe();
- this.primary.unsubscribe();
- }
- }
- }
-}
-
 export class GroupedObservable<K, T> extends Observable<T> {
  constructor(public key: K,
  private groupSubject: Subject<T>,
@@ -200,27 +174,28 @@ export class GroupedObservable<K, T> extends Observable<T> {
 
  protected _subscribe(subscriber: Subscriber<T>) {
  const subscription = new Subscription();
- if (this.refCountSubscription && !this.refCountSubscription.isUnsubscribed) {
- subscription.add(new InnerRefCountSubscription(this.refCountSubscription));
+ const {refCountSubscription, groupSubject} = this;
+ if (refCountSubscription && !refCountSubscription.isUnsubscribed) {
+ subscription.add(new InnerRefCountSubscription(refCountSubscription));
  }
- subscription.add(this.groupSubject.subscribe(subscriber));
+ subscription.add(groupSubject.subscribe(subscriber));
  return subscription;
  }
 }
 
-export class InnerRefCountSubscription extends Subscription {
+class InnerRefCountSubscription extends Subscription {
  constructor(private parent: RefCountSubscription) {
  super();
  parent.count++;
  }
 
  unsubscribe() {
- if (!this.parent.isUnsubscribed && !this.isUnsubscribed) {
+ const parent = this.parent;
+ if (!parent.isUnsubscribed && !this.isUnsubscribed) {
  super.unsubscribe();
- this.parent.count--;
- if (this.parent.count === 0 && this.parent.attemptedToUnsubscribePrimary) {
- this.parent.unsubscribe();
- this.parent.primary.unsubscribe();
+ parent.count -= 1;
+ if (parent.count === 0 && parent.attemptedToUnsubscribe) {
+ parent.unsubscribe();
  }
  }
  }