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'; /* tslint:disable:max-line-length */ export function groupBy(this: Observable, keySelector: (value: T) => K): Observable>; export function groupBy(this: Observable, keySelector: (value: T) => K, elementSelector: void, durationSelector: (grouped: GroupedObservable) => Observable): Observable>; export function groupBy(this: Observable, keySelector: (value: T) => K, elementSelector?: (value: T) => R, durationSelector?: (grouped: GroupedObservable) => Observable): Observable>; export function groupBy(this: Observable, keySelector: (value: T) => K, elementSelector?: (value: T) => R, durationSelector?: (grouped: GroupedObservable) => Observable, subjectSelector?: () => Subject): Observable>; /* tslint:disable:max-line-length */ /** * Groups the items emitted by an Observable according to a specified criterion, * and emits these grouped items as `GroupedObservables`, one * {@link GroupedObservable} per group. * * * * @param {function(value: T): K} keySelector a function that extracts the key * for each item. * @param {function(value: T): R} [elementSelector] a function that extracts the * return element for each item. * @param {function(grouped: GroupedObservable): Observable} [durationSelector] * a function that returns an Observable to determine how long each group should * exist. * @return {Observable>} an Observable that emits * GroupedObservables, each of which corresponds to a unique key value and each * of which emits those items from the source Observable that share that key * value. * @method groupBy * @owner Observable */ export function groupBy(this: Observable, keySelector: (value: T) => K, elementSelector?: ((value: T) => R) | void, durationSelector?: (grouped: GroupedObservable) => Observable, subjectSelector?: () => Subject): Observable> { return this.lift(new GroupByOperator(keySelector, elementSelector, durationSelector, subjectSelector)); } export interface RefCountSubscription { count: number; unsubscribe: () => void; closed: boolean; attemptedToUnsubscribe: boolean; } class GroupByOperator implements Operator> { constructor(private keySelector: (value: T) => K, private elementSelector?: ((value: T) => R) | void, private durationSelector?: (grouped: GroupedObservable) => Observable, private subjectSelector?: () => Subject) { } call(subscriber: Subscriber>, source: any): any { return source.subscribe(new GroupBySubscriber( subscriber, this.keySelector, this.elementSelector, this.durationSelector, this.subjectSelector )); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class GroupBySubscriber extends Subscriber implements RefCountSubscription { private groups: Map> = null; public attemptedToUnsubscribe: boolean = false; public count: number = 0; constructor(destination: Subscriber>, private keySelector: (value: T) => K, private elementSelector?: ((value: T) => R) | void, private durationSelector?: (grouped: GroupedObservable) => Observable, private subjectSelector?: () => Subject) { super(destination); } protected _next(value: T): void { let key: K; 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; if (!groups) { groups = this.groups = typeof key === 'string' ? new FastMap() : new Map(); } let group = groups.get(key); let element: R; if (this.elementSelector) { try { element = this.elementSelector(value); } catch (err) { this.error(err); } } else { element = value; } if (!group) { group = this.subjectSelector ? this.subjectSelector() : new Subject(); groups.set(key, group); const groupedObservable = new GroupedObservable(key, group, this); this.destination.next(groupedObservable); if (this.durationSelector) { let duration: any; try { duration = this.durationSelector(new GroupedObservable(key, >group)); } catch (err) { this.error(err); return; } this.add(duration.subscribe(new GroupDurationSubscriber(key, group, this))); } } if (!group.closed) { group.next(element); } } protected _error(err: any): void { const groups = this.groups; if (groups) { groups.forEach((group, key) => { group.error(err); }); groups.clear(); } this.destination.error(err); } protected _complete(): void { const groups = this.groups; if (groups) { groups.forEach((group, key) => { group.complete(); }); groups.clear(); } this.destination.complete(); } removeGroup(key: K): void { this.groups.delete(key); } unsubscribe() { if (!this.closed && !this.attemptedToUnsubscribe) { this.attemptedToUnsubscribe = true; if (this.count === 0) { super.unsubscribe(); } } } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class GroupDurationSubscriber extends Subscriber { constructor(private key: K, private group: Subject, private parent: GroupBySubscriber) { super(); } protected _next(value: T): void { this._complete(); } protected _error(err: any): void { const group = this.group; if (!group.closed) { group.error(err); } this.parent.removeGroup(this.key); } protected _complete(): void { const group = this.group; if (!group.closed) { group.complete(); } this.parent.removeGroup(this.key); } } /** * An Observable representing values belonging to the same group represented by * a common key. The values emitted by a GroupedObservable come from the source * Observable. The common key is available as the field `key` on a * GroupedObservable instance. * * @class GroupedObservable */ export class GroupedObservable extends Observable { constructor(public key: K, private groupSubject: Subject, private refCountSubscription?: RefCountSubscription) { super(); } protected _subscribe(subscriber: Subscriber) { const subscription = new Subscription(); const {refCountSubscription, groupSubject} = this; if (refCountSubscription && !refCountSubscription.closed) { subscription.add(new InnerRefCountSubscription(refCountSubscription)); } subscription.add(groupSubject.subscribe(subscriber)); return subscription; } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class InnerRefCountSubscription extends Subscription { constructor(private parent: RefCountSubscription) { super(); parent.count++; } unsubscribe() { const parent = this.parent; if (!parent.closed && !this.closed) { super.unsubscribe(); parent.count -= 1; if (parent.count === 0 && parent.attemptedToUnsubscribe) { parent.unsubscribe(); } } } }