Skip to content

Commit

Permalink
feat(groupBy): add higher-order lettable groupBy
Browse files Browse the repository at this point in the history
  • Loading branch information
@benlesh
benlesh committed Aug 9, 2017
1 parent 33eac1e commit 5281229
Show file tree
Hide file tree
Showing 4 changed files with 300 additions and 212 deletions.
2 changes: 1 addition & 1 deletion spec/operators/groupBy-spec.ts
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import {expect} from 'chai';
import * as Rx from '../../dist/cjs/Rx';
import {GroupedObservable} from '../../dist/cjs/operator/groupBy';
import {GroupedObservable} from '../../dist/cjs/operators/groupBy';
import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports

declare const { asDiagram };
Expand Down
214 changes: 3 additions & 211 deletions src/operator/groupBy.ts
View file
Original file line number Diff line number Diff line change
@@ -1,10 +1,7 @@
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 { groupBy as higherOrder, GroupedObservable } from '../operators/groupBy';

/* tslint:disable:max-line-length */
export function groupBy<T, K>(this: Observable<T>, keySelector: (value: T) => K): Observable<GroupedObservable<K, T>>;
Expand Down Expand Up @@ -84,210 +81,5 @@ export function groupBy<T, K, R>(this: Observable<T>, keySelector: (value: T) =>
elementSelector?: ((value: T) => R) | void,
durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>,
subjectSelector?: () => Subject<R>): Observable<GroupedObservable<K, R>> {
return this.lift(new GroupByOperator(keySelector, elementSelector, durationSelector, subjectSelector));
}

export interface RefCountSubscription {
count: number;
unsubscribe: () => void;
closed: boolean;
attemptedToUnsubscribe: boolean;
}

class GroupByOperator<T, K, R> implements Operator<T, GroupedObservable<K, R>> {
constructor(private keySelector: (value: T) => K,
private elementSelector?: ((value: T) => R) | void,
private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>,
private subjectSelector?: () => Subject<R>) {
}

call(subscriber: Subscriber<GroupedObservable<K, R>>, 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<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<GroupedObservable<K, R>>,
private keySelector: (value: T) => K,
private elementSelector?: ((value: T) => R) | void,
private durationSelector?: (grouped: GroupedObservable<K, R>) => Observable<any>,
private subjectSelector?: () => Subject<R>) {
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 = <any>value;
}

if (!group) {
group = this.subjectSelector ? this.subjectSelector() : new Subject<R>();
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<K, R>(key, <Subject<R>>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 = true;
if (this.count === 0) {
super.unsubscribe();
}
}
}
}

/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class GroupDurationSubscriber<K, T> extends Subscriber<T> {
constructor(private key: K,
private group: Subject<T>,
private parent: GroupBySubscriber<any, K, T>) {
super(group);
}

protected _next(value: T): void {
this.complete();
}

protected _unsubscribe() {
const { parent, key } = this;
this.key = this.parent = null;
if (parent) {
parent.removeGroup(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<K, T>
*/
export class GroupedObservable<K, T> extends Observable<T> {
constructor(public key: K,
private groupSubject: Subject<T>,
private refCountSubscription?: RefCountSubscription) {
super();
}

protected _subscribe(subscriber: Subscriber<T>) {
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();
}
}
}
return higherOrder(keySelector, elementSelector as any, durationSelector, subjectSelector)(this);
}
Loading

0 comments on commit 5281229

Please sign in to comment.