Skip to content

Commit 1e13aea

Browse files
benleshbenlesh
committed
feat(operator): add groupBy
closes #165
1 parent 7d9b52b commit 1e13aea

File tree

6 files changed

+286
-1
lines changed

6 files changed

+286
-1
lines changed

‎spec/operators/groupBy-spec.js‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
/* globals describe, it, expect */
2+
var Rx = require('../../dist/cjs/Rx');
3+
var Observable = Rx.Observable;
4+
5+
var noop = function () { };
6+
7+
describe('Observable.prototype.groupBy()', function () {
8+
9+
it('should group values', function (done) {
10+
var expectedGroups = [
11+
{ key: 1, values: [1, 3] },
12+
{ key: 0, values: [2] }
13+
];
14+
15+
Observable.of(1, 2, 3)
16+
.groupBy(function (x) { return x % 2 })
17+
.subscribe(function (g) {
18+
var expectedGroup = expectedGroups.shift();
19+
expect(g.key).toBe(expectedGroup.key);
20+
21+
g.subscribe(function (x) {
22+
expect(x).toBe(expectedGroup.values.shift());
23+
});
24+
}, null, done);
25+
});
26+
27+
it('should group values with an element selector', function (done) {
28+
var expectedGroups = [
29+
{ key: 1, values: ['1!', '3!'] },
30+
{ key: 0, values: ['2!'] }
31+
];
32+
33+
Observable.of(1, 2, 3)
34+
.groupBy(function (x) { return x % 2 }, function (x) { return x + '!'; })
35+
.subscribe(function (g) {
36+
var expectedGroup = expectedGroups.shift();
37+
expect(g.key).toBe(expectedGroup.key);
38+
39+
g.subscribe(function (x) {
40+
expect(x).toBe(expectedGroup.values.shift());
41+
});
42+
}, null, done);
43+
});
44+
45+
46+
it('should group values with a duration selector', function (done) {
47+
var expectedGroups = [
48+
{ key: 1, values: [1, 3] },
49+
{ key: 0, values: [2, 4] },
50+
{ key: 1, values: [5] },
51+
{ key: 0, values: [6] }
52+
];
53+
54+
Observable.of(1, 2, 3, 4, 5, 6)
55+
.groupBy(function (x) { return x % 2 }, function (x) { return x; }, function (g) { return g.skip(1); })
56+
.subscribe(function (g) {
57+
var expectedGroup = expectedGroups.shift();
58+
expect(g.key).toBe(expectedGroup.key);
59+
60+
g.subscribe(function (x) {
61+
expect(x).toBe(expectedGroup.values.shift());
62+
});
63+
}, null, done);
64+
});
65+
});

‎src/Observable.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import Subscriber from './Subscriber';
66
import Subscription from './Subscription';
77
import ConnectableObservable from './observables/ConnectableObservable';
88
// HACK: the Babel part of the build doesn't like this reference.
9+
import { GroupSubject } from './operators/groupBy';
910
// seems to put it in an infinite loop.
1011
//import Notification from './Notification';
1112

@@ -152,6 +153,8 @@ export default class Observable<T> {
152153

153154
catch: (selector: (err: any, source: Observable<T>, caught: Observable<any>) => Observable<any>) => Observable<T>;
154155
retryWhen: (notifier: (errors: Observable<any>) => Observable<any>) => Observable<T>;
156+
157+
groupBy: <T, R>(keySelector: (value:T) => string, durationSelector?: (group:GroupSubject<R>) => Observable<any>, elementSelector?: (value:T) => R) => Observable<R>;
155158

156159
finally: (ensure: () => void, thisArg?: any) => Observable<T>;
157-
}
160+
}

‎src/Rx.ts‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -151,6 +151,10 @@ import _finally from './operators/finally';
151151

152152
observableProto.finally = _finally;
153153

154+
import groupBy from './operators/groupBy';
155+
156+
observableProto.groupBy = groupBy;
157+
154158
export default {
155159
Subject,
156160
Scheduler,

‎src/operators/groupBy.ts‎

Lines changed: 139 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,139 @@
1+
import Operator from '../Operator';
2+
import Observer from '../Observer';
3+
import Subscriber from '../Subscriber';
4+
import Observable from '../Observable';
5+
import Subject from '../Subject';
6+
import Map from '../util/Map';
7+
import FastMap from '../util/FastMap';
8+
9+
import tryCatch from '../util/tryCatch';
10+
import {errorObject} from '../util/errorObject';
11+
import bindCallback from '../util/bindCallback';
12+
13+
export default function groupBy<T, R>(keySelector: (value: T) => string,
14+
elementSelector?: (value: T) => R,
15+
durationSelector?: (grouped: GroupSubject<R>) => Observable<any>): Observable<GroupSubject<R>>
16+
{
17+
return this.lift(new GroupByOperator<T, R>(keySelector, durationSelector, elementSelector));
18+
}
19+
20+
export class GroupByOperator<T, R> extends Operator<T, R> {
21+
constructor(private keySelector: (value: T) => string,
22+
private durationSelector?: (grouped: GroupSubject<R>) => Observable<any>,
23+
private elementSelector?: (value: T) => R) {
24+
super();
25+
}
26+
27+
call(observer: Observer<R>): Observer<T> {
28+
return new GroupBySubscriber<T, R>(observer, this.keySelector, this.durationSelector, this.elementSelector);
29+
}
30+
}
31+
32+
export class GroupBySubscriber<T, R> extends Subscriber<T> {
33+
private groups = null;
34+
35+
constructor(destination: Observer<R>, private keySelector: (value: T) => string,
36+
private durationSelector?: (grouped: GroupSubject<R>) => Observable<any>,
37+
private elementSelector?: (value: T) => R) {
38+
super(destination);
39+
}
40+
41+
_next(x: T) {
42+
let key = tryCatch(this.keySelector)(x);
43+
if(key === errorObject) {
44+
this.error(key.e);
45+
} else {
46+
let groups = this.groups;
47+
const elementSelector = this.elementSelector;
48+
const durationSelector = this.durationSelector;
49+
50+
if (!groups) {
51+
groups = this.groups = typeof key === 'string' ? new FastMap() : new Map();
52+
}
53+
54+
let group: GroupSubject<R> = groups.get(key);
55+
56+
if (!group) {
57+
groups.set(key, group = new GroupSubject(key));
58+
59+
if (durationSelector) {
60+
let duration = tryCatch(durationSelector)(group);
61+
if (duration === errorObject) {
62+
this.error(duration.e);
63+
} else {
64+
this.add(duration.subscribe(new GroupDurationSubscriber(group, this)));
65+
}
66+
}
67+
68+
this.destination.next(group);
69+
}
70+
71+
if (elementSelector) {
72+
let value = tryCatch(elementSelector)(x)
73+
if(value === errorObject) {
74+
group.error(value.e);
75+
} else {
76+
group.next(value);
77+
}
78+
} else {
79+
group.next(x);
80+
}
81+
}
82+
}
83+
84+
_error(err: any) {
85+
const groups = this.groups;
86+
if (groups) {
87+
groups.forEach((group, key) => {
88+
group.error(err);
89+
this.removeGroup(key);
90+
});
91+
}
92+
this.destination.error(err);
93+
}
94+
95+
_complete() {
96+
const groups = this.groups;
97+
if(groups) {
98+
groups.forEach((group, key) => {
99+
group.complete();
100+
this.removeGroup(group);
101+
});
102+
}
103+
this.destination.complete();
104+
}
105+
106+
removeGroup(key: string) {
107+
this.groups[key] = null;
108+
}
109+
}
110+
111+
export class GroupSubject<T> extends Subject<T> {
112+
constructor(public key: string) {
113+
super();
114+
}
115+
}
116+
117+
export class GroupDurationSubscriber<T> extends Subscriber<T> {
118+
constructor(private group: GroupSubject<T>, private parent:GroupBySubscriber<any, T>) {
119+
super(null);
120+
}
121+
122+
_next(value: T) {
123+
const group = this.group;
124+
group.complete();
125+
this.parent.removeGroup(group.key);
126+
}
127+
128+
_error(err: any) {
129+
const group = this.group;
130+
group.error(err);
131+
this.parent.removeGroup(group.key);
132+
}
133+
134+
_complete() {
135+
const group = this.group;
136+
group.complete();
137+
this.parent.removeGroup(group.key);
138+
}
139+
}

‎src/util/FastMap.ts‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
export default class FastMap {
2+
size: number = 0;
3+
private _values: Object = {};
4+
5+
delete(key: string): boolean {
6+
this._values[key] = null;
7+
return true;
8+
}
9+
10+
set(key: string, value: any): FastMap {
11+
this._values[key] = value;
12+
return this;
13+
}
14+
15+
get(key: string): any {
16+
return this._values[key];
17+
}
18+
19+
forEach(cb, thisArg) {
20+
const values = this._values;
21+
for (let key in values) {
22+
if (values.hasOwnProperty(key)) {
23+
cb.call(thisArg, values[key], key);
24+
}
25+
}
26+
}
27+
28+
clear() {
29+
this._values = {};
30+
}
31+
}

‎src/util/Map.ts‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
import { root } from './root';
2+
3+
export default root.Map || (function() {
4+
function Map() {
5+
this.size = 0;
6+
this._values = [];
7+
this._keys = [];
8+
}
9+
10+
Map.prototype['delete'] = function (key) {
11+
var i = this._keys.indexOf(key);
12+
if (i === -1) { return false }
13+
this._values.splice(i, 1);
14+
this._keys.splice(i, 1);
15+
this.size--;
16+
return true;
17+
};
18+
19+
Map.prototype.get = function (key) {
20+
var i = this._keys.indexOf(key);
21+
return i === -1 ? undefined : this._values[i];
22+
};
23+
24+
Map.prototype.set = function (key, value) {
25+
var i = this._keys.indexOf(key);
26+
if (i === -1) {
27+
this._keys.push(key);
28+
this._values.push(value);
29+
this.size++;
30+
} else {
31+
this._values[i] = value;
32+
}
33+
return this;
34+
};
35+
36+
Map.prototype.forEach = function (cb, thisArg) {
37+
for (var i = 0; i < this.size; i++) {
38+
cb.call(thisArg, this._values[i], this._keys[i]);
39+
}
40+
};
41+
42+
return Map;
43+
}());

0 commit comments

Comments
 (0)