Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Add buffer operator (#33)
* Add initial buffer implementation
* Add tests for buffer operator
authored by
Phil Plückthun
and committed by
GitHub
244a4269
afe433d8
+140
-2
+58
__tests__/wonka_test.re
+58
__tests__/wonka_test.re
···
1464
1464
)
1465
1465
);
1466
1466
});
1467
+
1468
+
describe("buffer", () => {
1469
+
open Expect;
1470
+
open! Expect.Operators;
1471
+
1472
+
beforeEach(() => Jest.useFakeTimers());
1473
+
afterEach(() => Jest.useRealTimers());
1474
+
1475
+
it("should buffer values and emit them on each notifier tick", () => {
1476
+
let a = Wonka.interval(50);
1477
+
1478
+
let talkback = ref((. _: Wonka_types.talkbackT) => ());
1479
+
let signals = [||];
1480
+
1481
+
let source = Wonka.buffer(Wonka.interval(100), a) |> Wonka.take(2);
1482
+
1483
+
source((. signal) =>
1484
+
switch (signal) {
1485
+
| Start(x) =>
1486
+
talkback := x;
1487
+
x(. Pull);
1488
+
| Push(_) =>
1489
+
ignore(Js.Array.push(signal, signals));
1490
+
talkback^(. Pull);
1491
+
| End => ignore(Js.Array.push(signal, signals))
1492
+
}
1493
+
);
1494
+
1495
+
Jest.runTimersToTime(400);
1496
+
1497
+
expect(signals) == [|Push([|0, 1|]), Push([|2, 3|]), End|];
1498
+
});
1499
+
1500
+
it("should end when the notifier ends", () => {
1501
+
let a = Wonka.interval(50) |> Wonka.take(3);
1502
+
1503
+
let talkback = ref((. _: Wonka_types.talkbackT) => ());
1504
+
let signals = [||];
1505
+
1506
+
let source = Wonka.buffer(Wonka.interval(100), a);
1507
+
1508
+
source((. signal) =>
1509
+
switch (signal) {
1510
+
| Start(x) =>
1511
+
talkback := x;
1512
+
x(. Pull);
1513
+
| Push(_) =>
1514
+
ignore(Js.Array.push(signal, signals));
1515
+
talkback^(. Pull);
1516
+
| End => ignore(Js.Array.push(signal, signals))
1517
+
}
1518
+
);
1519
+
1520
+
Jest.runTimersToTime(400);
1521
+
1522
+
expect(signals) == [|Push([|0, 1|]), Push([|2|]), End|];
1523
+
});
1524
+
});
1467
1525
});
1468
1526
1469
1527
describe("sink factories", () => {
+3
src/operators/wonka_operator_buffer.d.ts
+3
src/operators/wonka_operator_buffer.d.ts
+72
src/operators/wonka_operator_buffer.re
+72
src/operators/wonka_operator_buffer.re
···
1
+
open Wonka_types;
2
+
open Wonka_helpers;
3
+
4
+
type bufferStateT('a) = {
5
+
mutable buffer: Rebel.MutableQueue.t('a),
6
+
mutable sourceTalkback: (. talkbackT) => unit,
7
+
mutable notifierTalkback: (. talkbackT) => unit,
8
+
mutable ended: bool,
9
+
};
10
+
11
+
let buffer = (notifier: sourceT('a)) =>
12
+
curry((source: sourceT('b)) =>
13
+
curry((sink: sinkT(array('b))) => {
14
+
let state: bufferStateT('b) = {
15
+
buffer: Rebel.MutableQueue.make(),
16
+
sourceTalkback: talkbackPlaceholder,
17
+
notifierTalkback: talkbackPlaceholder,
18
+
ended: false,
19
+
};
20
+
21
+
source((. signal) =>
22
+
switch (signal) {
23
+
| Start(tb) =>
24
+
state.sourceTalkback = tb;
25
+
26
+
notifier((. signal) =>
27
+
switch (signal) {
28
+
| Start(tb) =>
29
+
state.notifierTalkback = tb;
30
+
state.notifierTalkback(. Pull);
31
+
| Push(_) when !state.ended =>
32
+
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
33
+
state.buffer = Rebel.MutableQueue.make();
34
+
state.notifierTalkback(. Pull);
35
+
| Push(_) => ()
36
+
| End when !state.ended =>
37
+
state.ended = true;
38
+
state.sourceTalkback(. Close);
39
+
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
40
+
sink(. End);
41
+
| End => ()
42
+
}
43
+
);
44
+
| Push(value) when !state.ended =>
45
+
Rebel.MutableQueue.add(state.buffer, value);
46
+
state.sourceTalkback(. Pull);
47
+
| Push(_) => ()
48
+
| End when !state.ended =>
49
+
state.ended = true;
50
+
state.notifierTalkback(. Close);
51
+
sink(. Push(Rebel.MutableQueue.toArray(state.buffer)));
52
+
sink(. End);
53
+
| End => ()
54
+
}
55
+
);
56
+
57
+
sink(.
58
+
Start(
59
+
(. signal) =>
60
+
if (!state.ended) {
61
+
switch (signal) {
62
+
| Close =>
63
+
state.ended = true;
64
+
state.sourceTalkback(. Close);
65
+
state.notifierTalkback(. Close);
66
+
| Pull => state.sourceTalkback(. Pull)
67
+
};
68
+
},
69
+
),
70
+
);
71
+
})
72
+
);
+3
src/operators/wonka_operator_buffer.rei
+3
src/operators/wonka_operator_buffer.rei
+2
-2
src/operators/wonka_operator_takeUntil.re
+2
-2
src/operators/wonka_operator_takeUntil.re
···
28
28
innerTb(. Pull);
29
29
| Push(_) =>
30
30
state.ended = true;
31
-
state.notifierTalkback(. Close);
32
31
state.sourceTalkback(. Close);
33
32
sink(. End);
34
33
| End => ()
35
34
}
36
35
);
37
36
| End when !state.ended =>
38
-
state.notifierTalkback(. Close);
39
37
state.ended = true;
38
+
state.notifierTalkback(. Close);
40
39
sink(. End);
41
40
| End => ()
42
41
| Push(_) when !state.ended => sink(. signal)
···
50
49
if (!state.ended) {
51
50
switch (signal) {
52
51
| Close =>
52
+
state.ended = true;
53
53
state.sourceTalkback(. Close);
54
54
state.notifierTalkback(. Close);
55
55
| Pull => state.sourceTalkback(. Pull)
+1
src/wonka.d.ts
+1
src/wonka.d.ts
···
7
7
export * from './sources/wonka_source_primitives';
8
8
9
9
/* operators */
10
+
export * from './operators/wonka_operator_buffer';
10
11
export * from './operators/wonka_operator_combine';
11
12
export * from './operators/wonka_operator_concatMap';
12
13
export * from './operators/wonka_operator_filter';