Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Fix edge cases of buffer operator (#51)
* Fix passesPassivePull for buffer operator
* Fix passesSinkClose and passesStrictEnd for buffer operator
* Fix passesActivePush and passesSourceEnd for buffer operator
authored by
Phil Plückthun
and committed by
GitHub
fdbba03f
b7c33ca0
+43
-22
+23
-10
src/wonka_operators.re
+23
-10
src/wonka_operators.re
···
5
5
mutable buffer: Rebel.MutableQueue.t('a),
6
6
mutable sourceTalkback: (. talkbackT) => unit,
7
7
mutable notifierTalkback: (. talkbackT) => unit,
8
+
mutable pulled: bool,
8
9
mutable ended: bool,
9
10
};
10
11
···
16
17
buffer: Rebel.MutableQueue.make(),
17
18
sourceTalkback: talkbackPlaceholder,
18
19
notifierTalkback: talkbackPlaceholder,
20
+
pulled: false,
19
21
ended: false,
20
22
};
21
23
···
26
28
27
29
notifier((. signal) =>
28
30
switch (signal) {
29
-
| Start(tb) =>
30
-
state.notifierTalkback = tb;
31
-
state.notifierTalkback(. Pull);
31
+
| Start(tb) => state.notifierTalkback = tb
32
32
| Push(_) when !state.ended =>
33
-
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
34
-
state.buffer = Rebel.MutableQueue.make();
35
-
state.notifierTalkback(. Pull);
33
+
if (Rebel.MutableQueue.size(state.buffer) > 0) {
34
+
let buffer = state.buffer;
35
+
state.buffer = Rebel.MutableQueue.make();
36
+
sink(. Push(Rebel.MutableQueue.toArray(buffer)));
37
+
}
36
38
| Push(_) => ()
37
39
| End when !state.ended =>
38
40
state.ended = true;
39
41
state.sourceTalkback(. Close);
40
-
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
42
+
if (Rebel.MutableQueue.size(state.buffer) > 0) {
43
+
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
44
+
};
41
45
sink(. End);
42
46
| End => ()
43
47
}
44
48
);
45
49
| Push(value) when !state.ended =>
46
50
Rebel.MutableQueue.add(state.buffer, value);
47
-
state.sourceTalkback(. Pull);
51
+
state.pulled = false;
52
+
state.notifierTalkback(. Pull);
48
53
| Push(_) => ()
49
54
| End when !state.ended =>
50
55
state.ended = true;
51
56
state.notifierTalkback(. Close);
52
-
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
57
+
if (Rebel.MutableQueue.size(state.buffer) > 0) {
58
+
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
59
+
};
53
60
sink(. End);
54
61
| End => ()
55
62
}
···
64
71
state.ended = true;
65
72
state.sourceTalkback(. Close);
66
73
state.notifierTalkback(. Close);
67
-
| Pull => state.sourceTalkback(. Pull)
74
+
| Pull =>
75
+
if (!state.pulled) {
76
+
state.pulled = true;
77
+
state.sourceTalkback(. Pull);
78
+
state.notifierTalkback(. Pull);
79
+
}
80
+
| Pull => ()
68
81
};
69
82
},
70
83
),
+20
-12
src/wonka_operators.test.ts
+20
-12
src/wonka_operators.test.ts
···
15
15
) =>
16
16
it('responds to Pull talkback signals (spec)', () => {
17
17
let talkback = null;
18
+
let push = 0;
18
19
const values = [];
19
20
20
21
const source: types.sourceT<any> = sink => {
21
22
sink(deriving.start(tb => {
22
-
if (tb === deriving.pull)
23
+
if (!push && tb === deriving.pull) {
24
+
push++;
23
25
sink(deriving.push(0));
26
+
}
24
27
}));
25
28
};
26
29
···
334
337
});
335
338
336
339
describe('buffer', () => {
337
-
const noop = operators.buffer(
338
-
operators.merge([
339
-
sources.fromValue(null),
340
-
sources.never
341
-
])
342
-
);
340
+
const valueThenNever: types.sourceT<any> = sink =>
341
+
sink(deriving.start(tb => {
342
+
if (tb === deriving.pull)
343
+
sink(deriving.push(null));
344
+
}));
343
345
344
-
// TODO: passesPassivePull(noop, [0]);
345
-
// TODO: passesActivePush(noop);
346
-
// TODO: passesSinkClose(noop);
347
-
// TODO: passesSourceEnd(noop);
346
+
const noop = operators.buffer(valueThenNever);
347
+
348
+
passesPassivePull(noop, [0]);
349
+
passesActivePush(noop, [0]);
350
+
passesSinkClose(noop);
351
+
passesSourceEnd(noop, [0]);
348
352
passesSingleStart(noop);
349
-
// TODO: passesStrictEnd(noop);
353
+
passesStrictEnd(noop);
350
354
351
355
it('emits batches of input values when a notifier emits', () => {
352
356
const { source: notifier$, next: notify } = sources.makeSubject();
···
361
365
362
366
notify(null);
363
367
expect(fn).toHaveBeenCalledWith([1, 2]);
368
+
369
+
next(3);
370
+
notify(null);
371
+
expect(fn).toHaveBeenCalledWith([3]);
364
372
});
365
373
});
366
374