Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
feat: Implement variadic zip and combine (#116)
* Reimplement combine as variadic function
* Implement zip & combine
* Fix up combine types
* Support objects in zip source
* Add additional combine test
* Move combine/zip tests to separate file
* Move size check in zip
* Fix up tuple mapping type
authored by
Phil Pluckthun
and committed by
GitHub
e08d998b
6881788c
+165
-102
+1
-1
package.json
+1
-1
package.json
+74
src/__tests__/combine.test.ts
+74
src/__tests__/combine.test.ts
···
1
+
import { Source } from '../types';
2
+
import { fromValue, makeSubject } from '../sources';
3
+
import { forEach } from '../sinks';
4
+
5
+
import {
6
+
passesPassivePull,
7
+
passesActivePush,
8
+
passesSinkClose,
9
+
passesSourceEnd,
10
+
passesSingleStart,
11
+
passesStrictEnd,
12
+
} from './compliance';
13
+
14
+
import { combine, zip } from '../combine';
15
+
16
+
beforeEach(() => {
17
+
jest.useFakeTimers();
18
+
});
19
+
20
+
describe('zip', () => {
21
+
const noop = (source: Source<any>) => zip([fromValue(0), source]);
22
+
23
+
passesPassivePull(noop, [0, 0]);
24
+
passesActivePush(noop, [0, 0]);
25
+
passesSinkClose(noop);
26
+
passesSourceEnd(noop, [0, 0]);
27
+
passesSingleStart(noop);
28
+
passesStrictEnd(noop);
29
+
30
+
it('emits the zipped values of two sources', () => {
31
+
const { source: sourceA, next: nextA } = makeSubject<number>();
32
+
const { source: sourceB, next: nextB } = makeSubject<number>();
33
+
const fn = jest.fn();
34
+
35
+
const combined = combine(sourceA, sourceB);
36
+
forEach(fn)(combined);
37
+
38
+
nextA(1);
39
+
expect(fn).not.toHaveBeenCalled();
40
+
nextB(2);
41
+
expect(fn).toHaveBeenCalledWith([1, 2]);
42
+
});
43
+
44
+
it('emits the zipped values of three sources', () => {
45
+
const { source: sourceA, next: nextA } = makeSubject<number>();
46
+
const { source: sourceB, next: nextB } = makeSubject<number>();
47
+
const { source: sourceC, next: nextC } = makeSubject<number>();
48
+
const fn = jest.fn();
49
+
50
+
const combined = zip([sourceA, sourceB, sourceC]);
51
+
forEach(fn)(combined);
52
+
53
+
nextA(1);
54
+
expect(fn).not.toHaveBeenCalled();
55
+
nextB(2);
56
+
expect(fn).not.toHaveBeenCalled();
57
+
nextC(3);
58
+
expect(fn).toHaveBeenCalledWith([1, 2, 3]);
59
+
});
60
+
61
+
it('emits the zipped values of a dictionary of two sources', () => {
62
+
const { source: sourceA, next: nextA } = makeSubject<number>();
63
+
const { source: sourceB, next: nextB } = makeSubject<number>();
64
+
const fn = jest.fn();
65
+
66
+
const combined = zip({ a: sourceA, b: sourceB });
67
+
forEach(fn)(combined);
68
+
69
+
nextA(1);
70
+
expect(fn).not.toHaveBeenCalled();
71
+
nextB(2);
72
+
expect(fn).toHaveBeenCalledWith({ a: 1, b: 2 });
73
+
});
74
+
});
-24
src/__tests__/operators.test.ts
-24
src/__tests__/operators.test.ts
···
21
21
jest.useFakeTimers();
22
22
});
23
23
24
-
describe('combine', () => {
25
-
const noop = (source: Source<any>) => operators.combine(sources.fromValue(0), source);
26
-
27
-
passesPassivePull(noop, [0, 0]);
28
-
passesActivePush(noop, [0, 0]);
29
-
passesSinkClose(noop);
30
-
passesSourceEnd(noop, [0, 0]);
31
-
passesSingleStart(noop);
32
-
passesStrictEnd(noop);
33
-
34
-
it('emits the zipped values of two sources', () => {
35
-
const { source: sourceA, next: nextA } = sources.makeSubject();
36
-
const { source: sourceB, next: nextB } = sources.makeSubject();
37
-
const fn = jest.fn();
38
-
39
-
sinks.forEach(fn)(operators.combine(sourceA, sourceB));
40
-
41
-
nextA(1);
42
-
expect(fn).not.toHaveBeenCalled();
43
-
nextB(2);
44
-
expect(fn).toHaveBeenCalledWith([1, 2]);
45
-
});
46
-
});
47
-
48
24
describe('buffer', () => {
49
25
const valueThenNever: Source<any> = sink =>
50
26
sink(
+82
src/combine.ts
+82
src/combine.ts
···
1
+
import { Source, TypeOfSource, SignalKind, TalkbackKind, TalkbackFn } from './types';
2
+
import { push, start, talkbackPlaceholder } from './helpers';
3
+
4
+
type TypeOfSourceArray<T extends readonly [...any[]]> = T extends [infer Head, ...infer Tail]
5
+
? [TypeOfSource<Head>, ...TypeOfSourceArray<Tail>]
6
+
: [];
7
+
8
+
export function zip<Sources extends readonly [...Source<any>[]]>(
9
+
sources: [...Sources]
10
+
): Source<TypeOfSourceArray<Sources>>;
11
+
12
+
export function zip<Sources extends { [prop: string]: Source<any> }>(
13
+
sources: Sources
14
+
): Source<{ [Property in keyof Sources]: TypeOfSource<Sources[Property]> }>;
15
+
16
+
export function zip<T>(
17
+
sources: Source<T>[] | Record<string, Source<T>>
18
+
): Source<T[] | Record<string, T>> {
19
+
const size = Object.keys(sources).length;
20
+
return sink => {
21
+
const filled: Set<string | number> = new Set();
22
+
23
+
const talkbacks: TalkbackFn[] | Record<string, TalkbackFn | void> = Array.isArray(sources)
24
+
? new Array(size).fill(talkbackPlaceholder)
25
+
: {};
26
+
const buffer: T[] | Record<string, T> = Array.isArray(sources) ? new Array(size) : {};
27
+
28
+
let gotBuffer = false;
29
+
let gotSignal = false;
30
+
let ended = false;
31
+
let endCount = 0;
32
+
33
+
for (const key in sources) {
34
+
(sources[key] as Source<T>)(signal => {
35
+
if (signal === SignalKind.End) {
36
+
if (endCount >= size - 1) {
37
+
ended = true;
38
+
sink(SignalKind.End);
39
+
} else {
40
+
endCount++;
41
+
}
42
+
} else if (signal.tag === SignalKind.Start) {
43
+
talkbacks[key] = signal[0];
44
+
} else if (!ended) {
45
+
buffer[key] = signal[0];
46
+
filled.add(key);
47
+
if (!gotBuffer && filled.size < size) {
48
+
if (!gotSignal) {
49
+
for (const key in sources)
50
+
if (!filled.has(key)) (talkbacks[key] || talkbackPlaceholder)(TalkbackKind.Pull);
51
+
} else {
52
+
gotSignal = false;
53
+
}
54
+
} else {
55
+
gotBuffer = true;
56
+
gotSignal = false;
57
+
sink(push(Array.isArray(buffer) ? buffer.slice() : { ...buffer }));
58
+
}
59
+
}
60
+
});
61
+
}
62
+
sink(
63
+
start(signal => {
64
+
if (ended) {
65
+
/*noop*/
66
+
} else if (signal === TalkbackKind.Close) {
67
+
ended = true;
68
+
for (const key in talkbacks) talkbacks[key](TalkbackKind.Close);
69
+
} else if (!gotSignal) {
70
+
gotSignal = true;
71
+
for (const key in talkbacks) talkbacks[key](TalkbackKind.Pull);
72
+
}
73
+
})
74
+
);
75
+
};
76
+
}
77
+
78
+
export function combine<Sources extends Source<any>[]>(
79
+
...sources: Sources
80
+
): Source<TypeOfSourceArray<Sources>> {
81
+
return zip(sources);
82
+
}
+1
src/index.ts
+1
src/index.ts
-73
src/operators.ts
-73
src/operators.ts
···
64
64
};
65
65
}
66
66
67
-
export function combine<A, B>(sourceA: Source<A>, sourceB: Source<B>): Source<[A, B]> {
68
-
return sink => {
69
-
let lastValA: A | void;
70
-
let lastValB: B | void;
71
-
let talkbackA = talkbackPlaceholder;
72
-
let talkbackB = talkbackPlaceholder;
73
-
let gotSignal = false;
74
-
let gotEnd = false;
75
-
let ended = false;
76
-
sourceA(signal => {
77
-
if (signal === SignalKind.End) {
78
-
if (!gotEnd) {
79
-
gotEnd = true;
80
-
} else {
81
-
ended = true;
82
-
sink(SignalKind.End);
83
-
}
84
-
} else if (signal.tag === SignalKind.Start) {
85
-
talkbackA = signal[0];
86
-
} else if (lastValB === undefined) {
87
-
lastValA = signal[0];
88
-
if (!gotSignal) {
89
-
talkbackB(TalkbackKind.Pull);
90
-
} else {
91
-
gotSignal = false;
92
-
}
93
-
} else if (!ended) {
94
-
lastValA = signal[0];
95
-
gotSignal = false;
96
-
sink(push([lastValA, lastValB] as [A, B]));
97
-
}
98
-
});
99
-
sourceB(signal => {
100
-
if (signal === SignalKind.End) {
101
-
if (!gotEnd) {
102
-
gotEnd = true;
103
-
} else {
104
-
ended = true;
105
-
sink(SignalKind.End);
106
-
}
107
-
} else if (signal.tag === SignalKind.Start) {
108
-
talkbackB = signal[0];
109
-
} else if (lastValA === undefined) {
110
-
lastValB = signal[0];
111
-
if (!gotSignal) {
112
-
talkbackA(TalkbackKind.Pull);
113
-
} else {
114
-
gotSignal = false;
115
-
}
116
-
} else if (!ended) {
117
-
lastValB = signal[0];
118
-
gotSignal = false;
119
-
sink(push([lastValA, lastValB] as [A, B]));
120
-
}
121
-
});
122
-
sink(
123
-
start(signal => {
124
-
if (ended) {
125
-
/*noop*/
126
-
} else if (signal === TalkbackKind.Close) {
127
-
ended = true;
128
-
talkbackA(TalkbackKind.Close);
129
-
talkbackB(TalkbackKind.Close);
130
-
} else if (!gotSignal) {
131
-
gotSignal = true;
132
-
talkbackA(TalkbackKind.Pull);
133
-
talkbackB(TalkbackKind.Pull);
134
-
}
135
-
})
136
-
);
137
-
};
138
-
}
139
-
140
67
export function concatMap<In, Out>(map: (value: In) => Source<Out>): Operator<In, Out> {
141
68
return source => sink => {
142
69
const inputQueue: In[] = [];
+3
src/types.ts
+3
src/types.ts
···
33
33
/** An operator transforms a [Source] and returns a new [Source], potentially with different timings or output types. */
34
34
export type Operator<In, Out> = (a: Source<In>) => Source<Out>;
35
35
36
+
/** Extracts the type of a given Source */
37
+
export type TypeOfSource<T> = T extends Source<infer U> ? U : never;
38
+
36
39
export interface Subscription {
37
40
unsubscribe(): void;
38
41
}
+4
-4
yarn.lock
+4
-4
yarn.lock
···
3890
3890
resolved "https://registry.yarnpkg.com/typescript-compiler/-/typescript-compiler-1.4.1-2.tgz#ba4f7db22d91534a1929d90009dce161eb72fd3f"
3891
3891
integrity sha512-EMopKmoAEJqA4XXRFGOb7eSBhmQMbBahW6P1Koayeatp0b4AW2q/bBqYWkpG7QVQc9HGQUiS4trx2ZHcnAaZUg==
3892
3892
3893
-
typescript@^4.7.4:
3894
-
version "4.7.4"
3895
-
resolved "https://registry.yarnpkg.com/typescript/-/typescript-4.7.4.tgz#1a88596d1cf47d59507a1bcdfb5b9dfe4d488235"
3896
-
integrity sha512-C0WQT0gezHuw6AdY1M2jxUO83Rjf0HP7Sk1DtXj6j1EwkQNZrHAg2XPWlq62oqEhYvONq5pkC2Y9oPljWToLmQ==
3893
+
typescript@^4.8.2:
3894
+
version "4.8.2"
3895
+
resolved "https://registry.yarnpkg.com/typescript/-/typescript-4.8.2.tgz#e3b33d5ccfb5914e4eeab6699cf208adee3fd790"
3896
+
integrity sha512-C0I1UsrrDHo2fYI5oaCGbSejwX4ch+9Y5jTQELvovfmFkK3HHSZJB8MSJcWLmCUBzQBchCrZ9rMRV6GuNrvGtw==
3897
3897
3898
3898
typescript@~4.4.4:
3899
3899
version "4.4.4"