Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Implement toArray sink (#35)
* Add initial toArray sink implementation
* Add tests for toArray source
* Modify toArray to match subscribe behaviour
* Close non-synchronous sources in toArray
* Add additional toArray test for async sources
* Add docs for toArray sink
* Refactor makeTrampoline and use it for fromArray and fromList
authored by
Phil Plückthun
and committed by
GitHub
5787fc94
39bb9cef
+174
-99
+39
__tests__/wonka_test.re
+39
__tests__/wonka_test.re
···
1692
1692
})
1693
1693
)
1694
1694
);
1695
+
1696
+
describe("toArray", () => {
1697
+
open Expect;
1698
+
1699
+
it("converts iterable sources to arrays", () => {
1700
+
let input = [|1, 2, 3|];
1701
+
let output = Wonka.fromArray(input) |> Wonka.toArray;
1702
+
expect(output) |> toEqual(input);
1703
+
});
1704
+
1705
+
it("converts mapped iterable sources to arrays", () => {
1706
+
let input = [|1, 2, 3|];
1707
+
let output =
1708
+
Wonka.fromArray(input) |> Wonka.map((. x) => x) |> Wonka.toArray;
1709
+
expect(output) |> toEqual(input);
1710
+
});
1711
+
1712
+
it("converts concatenated iterable sources to arrays", () => {
1713
+
let inputA = [|1, 2, 3|];
1714
+
let inputB = [|1, 2, 3|];
1715
+
let input = Array.append(inputA, inputB);
1716
+
1717
+
let output =
1718
+
Wonka.concat([|Wonka.fromArray(inputA), Wonka.fromArray(inputB)|])
1719
+
|> Wonka.toArray;
1720
+
1721
+
expect(output) |> toEqual(input);
1722
+
});
1723
+
1724
+
it("ignores and closes asynchronous push streams", () => {
1725
+
let input = [|1, 2, 3|];
1726
+
let source = Wonka.concat([|Wonka.fromArray(input), Wonka.never|]);
1727
+
let (signals, wrappedSource) =
1728
+
source |> Wonka_thelpers.testSourceOperator;
1729
+
1730
+
ignore(Wonka.toArray(wrappedSource));
1731
+
expect(signals) |> toEqual([|Pull, Close|]);
1732
+
});
1733
+
});
1695
1734
});
1696
1735
1697
1736
describe("chains (integration)", () =>
+22
__tests__/wonka_thelpers.re
+22
__tests__/wonka_thelpers.re
···
93
93
);
94
94
};
95
95
96
+
let testSourceOperator = source => {
97
+
let res = [||];
98
+
let innerSource = sink => {
99
+
source((. signal) =>
100
+
switch (signal) {
101
+
| Start(outerTalkback) =>
102
+
sink(.
103
+
Start(
104
+
(. talkback) => {
105
+
Js.Array.push(talkback, res);
106
+
outerTalkback(. talkback);
107
+
},
108
+
),
109
+
)
110
+
| _ => sink(. signal)
111
+
}
112
+
);
113
+
};
114
+
115
+
(res, innerSource);
116
+
};
117
+
96
118
type observableClassT;
97
119
98
120
[@bs.module] external observableClass: observableClassT = "zen-observable";
+29
docs/api/sinks.md
+29
docs/api/sinks.md
···
92
92
); // Prints 123 to the console.
93
93
```
94
94
95
+
## toArray
96
+
97
+
`toArray` returns an array, which contains all values from a pull source.
98
+
This sink is primarily intended for synchronous pull streams. Passing it
99
+
an asynchronous push streams may result in an empty array being returned.
100
+
101
+
If you're passing an asynchronous push stream `toArray` will cancel it
102
+
before it returns an array.
103
+
104
+
> _Note:_ If you're using this sink, make sure that your input source streams
105
+
> the values you're collecting partly or fully synchronously.
106
+
107
+
```reason
108
+
Wonka.fromArray([|1, 2, 3|])
109
+
|> Wonka.map((. x) => x * 2)
110
+
|> Wonka.toArray
111
+
/* Returns [|2, 4, 6|] */
112
+
```
113
+
114
+
```typescript
115
+
import { pipe, fromArray, map, toArray } from 'wonka';
116
+
117
+
pipe(
118
+
fromArray([1, 2, 3]),
119
+
map(x => x * 2),
120
+
toArray
121
+
); // Returns [2, 4, 6]
122
+
```
123
+
95
124
## toPromise
96
125
97
126
`toPromise` returns a promise, which resolves on the last value of a source.
+3
src/sinks/wonka_sink_toArray.d.ts
+3
src/sinks/wonka_sink_toArray.d.ts
+36
src/sinks/wonka_sink_toArray.re
+36
src/sinks/wonka_sink_toArray.re
···
1
+
open Wonka_types;
2
+
open Wonka_helpers;
3
+
4
+
type toArrayStateT('a) = {
5
+
values: Rebel.MutableQueue.t('a),
6
+
mutable talkback: (. talkbackT) => unit,
7
+
mutable value: option('a),
8
+
mutable ended: bool,
9
+
};
10
+
11
+
let toArray = (source: sourceT('a)): array('a) => {
12
+
let state: toArrayStateT('a) = {
13
+
values: Rebel.MutableQueue.make(),
14
+
talkback: talkbackPlaceholder,
15
+
value: None,
16
+
ended: false,
17
+
};
18
+
19
+
source((. signal) =>
20
+
switch (signal) {
21
+
| Start(x) =>
22
+
state.talkback = x;
23
+
x(. Pull);
24
+
| Push(value) =>
25
+
Rebel.MutableQueue.add(state.values, value);
26
+
state.talkback(. Pull);
27
+
| End => state.ended = true
28
+
}
29
+
);
30
+
31
+
if (!state.ended) {
32
+
state.talkback(. Close);
33
+
};
34
+
35
+
Rebel.MutableQueue.toArray(state.values);
36
+
};
+3
src/sinks/wonka_sink_toArray.rei
+3
src/sinks/wonka_sink_toArray.rei
+10
-34
src/sources/wonka_source_fromArray.re
+10
-34
src/sources/wonka_source_fromArray.re
···
1
1
open Wonka_types;
2
-
3
-
type fromArrayState('a) = {
4
-
mutable index: int,
5
-
mutable ended: bool,
6
-
mutable looping: bool,
7
-
mutable pull: bool,
8
-
};
2
+
open Wonka_helpers;
9
3
10
4
let fromArray = arr =>
11
5
curry(sink => {
12
6
let size = Rebel.Array.size(arr);
13
-
let state = {index: 0, ended: false, looping: false, pull: false};
14
-
15
-
sink(.
16
-
Start(
17
-
(. signal) =>
18
-
switch (signal, state.looping) {
19
-
| (Pull, false) =>
20
-
state.pull = true;
21
-
state.looping = true;
22
-
23
-
while (state.pull && !state.ended) {
24
-
let index = state.index;
25
-
if (index < size) {
26
-
let x = Rebel.Array.getUnsafe(arr, index);
27
-
state.index = index + 1;
28
-
state.pull = false;
29
-
sink(. Push(x));
30
-
} else {
31
-
state.ended = true;
32
-
sink(. End);
33
-
};
34
-
};
7
+
let index = ref(0);
35
8
36
-
state.looping = false;
37
-
| (Pull, true) => state.pull = true
38
-
| (Close, _) => state.ended = true
39
-
},
40
-
),
9
+
makeTrampoline(sink, (.) =>
10
+
if (index^ < size) {
11
+
let x = Rebel.Array.getUnsafe(arr, index^);
12
+
index := index^ + 1;
13
+
Some(x);
14
+
} else {
15
+
None;
16
+
}
41
17
);
42
18
});
+9
-33
src/sources/wonka_source_fromList.re
+9
-33
src/sources/wonka_source_fromList.re
···
1
1
open Wonka_types;
2
-
3
-
type fromListState('a) = {
4
-
mutable value: 'a,
5
-
mutable ended: bool,
6
-
mutable looping: bool,
7
-
mutable pull: bool,
8
-
};
2
+
open Wonka_helpers;
9
3
10
4
let fromList = ls =>
11
5
curry(sink => {
12
-
let state = {value: ls, ended: false, looping: false, pull: false};
13
-
14
-
sink(.
15
-
Start(
16
-
(. signal) =>
17
-
switch (signal, state.looping) {
18
-
| (Pull, false) =>
19
-
state.pull = true;
20
-
state.looping = true;
21
-
22
-
while (state.pull && !state.ended) {
23
-
switch (state.value) {
24
-
| [x, ...rest] =>
25
-
state.value = rest;
26
-
state.pull = false;
27
-
sink(. Push(x));
28
-
| [] =>
29
-
state.ended = true;
30
-
sink(. End);
31
-
};
32
-
};
6
+
let value = ref(ls);
33
7
34
-
state.looping = false;
35
-
| (Pull, true) => state.pull = true
36
-
| (Close, _) => state.ended = true
37
-
},
38
-
),
8
+
makeTrampoline(sink, (.) =>
9
+
switch (value^) {
10
+
| [x, ...rest] =>
11
+
value := rest;
12
+
Some(x);
13
+
| [] => None
14
+
}
39
15
);
40
16
});
+1
src/wonka.d.ts
+1
src/wonka.d.ts
+21
-32
src/wonka_helpers.re
+21
-32
src/wonka_helpers.re
···
20
20
};
21
21
22
22
type trampolineT = {
23
-
mutable exhausted: bool,
24
-
mutable inLoop: bool,
25
-
mutable gotSignal: bool,
23
+
mutable ended: bool,
24
+
mutable looping: bool,
25
+
mutable pull: bool,
26
26
};
27
27
28
28
let makeTrampoline = (sink: sinkT('a), f: (. unit) => option('a)) => {
29
-
let state: trampolineT = {
30
-
exhausted: false,
31
-
inLoop: false,
32
-
gotSignal: false,
33
-
};
34
-
35
-
let loop = () => {
36
-
let rec explode = () =>
37
-
switch (f(.)) {
38
-
| Some(x) =>
39
-
state.gotSignal = false;
40
-
sink(. Push(x));
41
-
if (state.gotSignal) {
42
-
explode();
43
-
};
44
-
| None =>
45
-
state.exhausted = true;
46
-
sink(. End);
47
-
};
48
-
49
-
state.inLoop = true;
50
-
explode();
51
-
state.inLoop = false;
52
-
};
29
+
let state: trampolineT = {ended: false, looping: false, pull: false};
53
30
54
31
sink(.
55
32
Start(
56
33
(. signal) =>
57
-
switch (signal, state.exhausted) {
34
+
switch (signal, state.looping) {
58
35
| (Pull, false) =>
59
-
state.gotSignal = true;
60
-
if (!state.inLoop) {
61
-
loop();
36
+
state.pull = true;
37
+
state.looping = true;
38
+
39
+
while (state.pull && !state.ended) {
40
+
switch (f(.)) {
41
+
| Some(x) =>
42
+
state.pull = false;
43
+
sink(. Push(x));
44
+
| None =>
45
+
state.ended = true;
46
+
sink(. End);
47
+
};
62
48
};
63
-
| _ => ()
49
+
50
+
state.looping = false;
51
+
| (Pull, true) => state.pull = true
52
+
| (Close, _) => state.ended = true
64
53
},
65
54
),
66
55
);