Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Add {from,to}Observable and {from,to}Callbag (#31)
* Implement fromObservable operator
* Add toObservable and refactor
* Rename symbol to observableSymbol
* Support Observables without Symbol.observable
* Add fromCallbag and toCallbag
* Support synchronous observables
* Add tests for fromObservable and fromCallbag
* Fix Observable.symbol creation and toObservable
* Fix method usage on Symbol.observable
`this` was not respected since the method was separated
from the input observable.
* Add tests for toObservable and toCallbag
* Revert changes to fromObservable
authored by
Phil Plückthun
and committed by
GitHub
afe433d8
5d03235f
+386
-1
+83
__tests__/wonka_test.re
+83
__tests__/wonka_test.re
···
147
147
expect(ended^) === false;
148
148
});
149
149
});
150
+
151
+
describe("fromObservable", () =>
152
+
Expect.(
153
+
testPromise("creates a source from an observable", () => {
154
+
let observable = Wonka_thelpers.observableFromArray([|1, 2|]);
155
+
156
+
Wonka_thelpers.testSource(Wonka.fromObservable(observable))
157
+
|> Js.Promise.then_(x =>
158
+
expect(x)
159
+
|> toEqual([|Push(1), Push(2), End|])
160
+
|> Js.Promise.resolve
161
+
);
162
+
})
163
+
)
164
+
);
165
+
166
+
describe("fromCallbag", () => {
167
+
open Expect;
168
+
169
+
testPromise("creates a source from a callbag (observable)", () => {
170
+
let observable = Wonka_thelpers.observableFromArray([|1, 2|]);
171
+
let callbag = Wonka_thelpers.callbagFromObservable(observable);
172
+
173
+
Wonka_thelpers.testSource(Wonka.fromCallbag(callbag))
174
+
|> Js.Promise.then_(x =>
175
+
expect(x)
176
+
|> toEqual([|Push(1), Push(2), End|])
177
+
|> Js.Promise.resolve
178
+
);
179
+
});
180
+
181
+
testPromise("creates a source from a callbag (iterable)", () => {
182
+
let callbag = Wonka_thelpers.callbagFromArray([|1, 2|]);
183
+
184
+
Wonka_thelpers.testSource(Wonka.fromCallbag(callbag))
185
+
|> Js.Promise.then_(x =>
186
+
expect(x)
187
+
|> toEqual([|Push(1), Push(2), End|])
188
+
|> Js.Promise.resolve
189
+
);
190
+
});
191
+
});
150
192
});
151
193
152
194
describe("operator factories", () => {
···
1490
1532
expect(nums) |> toEqual([|0, 1|]);
1491
1533
},
1492
1534
)
1535
+
)
1536
+
);
1537
+
1538
+
describe("toObservable", () =>
1539
+
Expect.(
1540
+
testPromise("should convert a source to an Observable", () => {
1541
+
let source = Wonka.fromArray([|1, 2|]);
1542
+
let observable =
1543
+
Wonka.toObservable(source) |> Wonka_thelpers.observableFrom;
1544
+
let values = [||];
1545
+
1546
+
let promise =
1547
+
observable->Wonka_thelpers.observableForEach(value =>
1548
+
ignore(Js.Array.push(value, values))
1549
+
);
1550
+
1551
+
promise
1552
+
|> Js.Promise.then_(() =>
1553
+
expect(values) |> toEqual([|1, 2|]) |> Js.Promise.resolve
1554
+
);
1555
+
})
1556
+
)
1557
+
);
1558
+
1559
+
describe("toCallbag", () =>
1560
+
Expect.(
1561
+
it("should convert a source to a Callbag", () => {
1562
+
let source = Wonka.fromArray([|1, 2|]);
1563
+
let callbag = Wonka.toCallbag(source);
1564
+
let values = [||];
1565
+
1566
+
(
1567
+
Wonka_thelpers.callbagIterate(. value =>
1568
+
ignore(Js.Array.push(value, values))
1569
+
)
1570
+
)(.
1571
+
callbag,
1572
+
);
1573
+
1574
+
expect(values) |> toEqual([|1, 2|]);
1575
+
})
1493
1576
)
1494
1577
);
1495
1578
});
+54
__tests__/wonka_thelpers.re
+54
__tests__/wonka_thelpers.re
···
71
71
);
72
72
});
73
73
};
74
+
75
+
let testSource = source => {
76
+
let talkback = ref((. _: talkbackT) => ());
77
+
let res = [||];
78
+
79
+
Js.Promise.make((~resolve, ~reject as _) =>
80
+
source((. signal) =>
81
+
switch (signal) {
82
+
| Start(x) =>
83
+
talkback := x;
84
+
talkback^(. Pull);
85
+
| Push(_) =>
86
+
ignore(Js.Array.push(signal, res));
87
+
talkback^(. Pull);
88
+
| End =>
89
+
ignore(Js.Array.push(signal, res));
90
+
resolve(. res);
91
+
}
92
+
)
93
+
);
94
+
};
95
+
96
+
type observableClassT;
97
+
98
+
[@bs.module] external observableClass: observableClassT = "zen-observable";
99
+
[@bs.send]
100
+
external _observableFromArray:
101
+
(observableClassT, array('a)) => Wonka.observableT('a) =
102
+
"from";
103
+
[@bs.send]
104
+
external _observableFrom:
105
+
(observableClassT, Wonka.observableT('a)) => Wonka.observableT('a) =
106
+
"from";
107
+
[@bs.send]
108
+
external observableForEach:
109
+
(Wonka.observableT('a), 'a => unit) => Js.Promise.t(unit) =
110
+
"forEach";
111
+
112
+
let observableFromArray = (arr: array('a)): Wonka.observableT('a) =>
113
+
_observableFromArray(observableClass, arr);
114
+
let observableFrom = (obs: Wonka.observableT('a)): Wonka.observableT('a) =>
115
+
_observableFrom(observableClass, obs);
116
+
117
+
[@bs.module]
118
+
external callbagFromArray: array('a) => Wonka.callbagT('a) =
119
+
"callbag-from-iter";
120
+
121
+
[@bs.module]
122
+
external callbagFromObservable: Wonka.observableT('a) => Wonka.callbagT('a) =
123
+
"callbag-from-obs";
124
+
125
+
[@bs.module]
126
+
external callbagIterate: (. ('a => unit)) => (. Wonka.callbagT('a)) => unit =
127
+
"callbag-iterate";
+5
-1
package.json
+5
-1
package.json
···
51
51
"devDependencies": {
52
52
"@glennsl/bs-jest": "^0.4.8",
53
53
"bs-platform": "^5.0.4",
54
+
"callbag-from-iter": "^1.2.0",
55
+
"callbag-from-obs": "^1.2.0",
56
+
"callbag-iterate": "^1.0.0",
54
57
"codecov": "^3.5.0",
55
58
"flowgen": "^1.8.0",
56
59
"gatsby": "^2.11.0",
···
68
71
"rollup-plugin-commonjs": "^9.3.4",
69
72
"rollup-plugin-node-resolve": "^4.2.3",
70
73
"rollup-plugin-prettier": "^0.6.0",
71
-
"rollup-plugin-terser": "^4.0.4"
74
+
"rollup-plugin-terser": "^4.0.4",
75
+
"zen-observable": "^0.8.14"
72
76
},
73
77
"lint-staged": {
74
78
"*.{d.ts,js}": [
+4
src/web/wonkaJs.d.ts
+4
src/web/wonkaJs.d.ts
+4
src/web/wonkaJs.re
+4
src/web/wonkaJs.re
+10
src/web/wonka_callbag.d.ts
+10
src/web/wonka_callbag.d.ts
···
1
+
import { Source } from '../wonka_types';
2
+
3
+
export type Callbag<I, O> = {
4
+
(t: 0, d: Callbag<O, I>): void;
5
+
(t: 1, d: I): void;
6
+
(t: 2, d?: any): void;
7
+
};
8
+
9
+
export const fromCallbag: <T>(callbag: Callbag<void, T>) => Source<T>;
10
+
export const toCallbag: <T>(source: Source<T>) => Callbag<void, T>;
+57
src/web/wonka_callbag.re
+57
src/web/wonka_callbag.re
···
1
+
open Wonka_types;
2
+
3
+
type callbagSignal =
4
+
| CALLBAG_START /* 0 */
5
+
| CALLBAG_DATA /* 1 */
6
+
| CALLBAG_END /* 2 */;
7
+
8
+
type callbagData('a);
9
+
type callbagTalkback = (. callbagSignal) => unit;
10
+
type callbagT('a) = (. callbagSignal, callbagData('a)) => unit;
11
+
12
+
external unsafe_getCallbag: callbagData('a) => callbagT('a) = "%identity";
13
+
external unsafe_getTalkback: callbagData('a) => callbagTalkback = "%identity";
14
+
external unsafe_getValue: callbagData('a) => 'a = "%identity";
15
+
external unsafe_wrap: 'any => callbagData('a) = "%identity";
16
+
17
+
let fromCallbag = callbag =>
18
+
curry(sink => {
19
+
let wrappedSink =
20
+
(. signal, data) =>
21
+
switch (signal) {
22
+
| CALLBAG_START =>
23
+
let talkback = unsafe_getTalkback(data);
24
+
let wrappedTalkback = (
25
+
(. talkbackSignal: talkbackT) =>
26
+
switch (talkbackSignal) {
27
+
| Pull => talkback(. CALLBAG_DATA)
28
+
| Close => talkback(. CALLBAG_END)
29
+
}
30
+
);
31
+
sink(. Start(wrappedTalkback));
32
+
| CALLBAG_DATA => sink(. Push(unsafe_getValue(data)))
33
+
| CALLBAG_END => sink(. End)
34
+
};
35
+
callbag(. CALLBAG_START, unsafe_wrap(wrappedSink));
36
+
});
37
+
38
+
let toCallbag = source =>
39
+
curry((. signal, data) =>
40
+
if (signal === CALLBAG_START) {
41
+
let callbag = unsafe_getCallbag(data);
42
+
source((. signal) =>
43
+
switch (signal) {
44
+
| Start(talkbackFn) =>
45
+
let wrappedTalkbackFn = (talkback: callbagSignal) =>
46
+
switch (talkback) {
47
+
| CALLBAG_START => ()
48
+
| CALLBAG_DATA => talkbackFn(. Pull)
49
+
| CALLBAG_END => talkbackFn(. Close)
50
+
};
51
+
callbag(. CALLBAG_START, unsafe_wrap(wrappedTalkbackFn));
52
+
| Push(data) => callbag(. CALLBAG_DATA, unsafe_wrap(data))
53
+
| End => callbag(. CALLBAG_END, unsafe_wrap())
54
+
}
55
+
);
56
+
}
57
+
);
+6
src/web/wonka_callbag.rei
+6
src/web/wonka_callbag.rei
+18
src/web/wonka_observable.d.ts
+18
src/web/wonka_observable.d.ts
···
1
+
import { Source } from '../wonka_types';
2
+
3
+
export interface Subscription {
4
+
unsubscribe(): void;
5
+
}
6
+
7
+
export interface Observer<T> {
8
+
next(value: T): void;
9
+
error(errorValue: any): void;
10
+
complete(): void;
11
+
}
12
+
13
+
export interface Observable<T> {
14
+
subscribe(observer: Observer<T>): Subscription;
15
+
}
16
+
17
+
export const fromObservable: <T>(observable: Observable<T>) => Source<T>;
18
+
export const toObservable: <T>(source: Source<T>) => Observable<T>;
+117
src/web/wonka_observable.re
+117
src/web/wonka_observable.re
···
1
+
open Wonka_types;
2
+
open Wonka_helpers;
3
+
4
+
let observableSymbol: string = [%raw
5
+
{|
6
+
typeof Symbol === 'function'
7
+
? Symbol.observable || (Symbol.observable = Symbol('observable'))
8
+
: '@@observable'
9
+
|}
10
+
];
11
+
12
+
type subscriptionT = {. [@bs.meth] "unsubscribe": unit => unit};
13
+
14
+
type observerT('a) = {
15
+
.
16
+
[@bs.meth] "next": 'a => unit,
17
+
[@bs.meth] "error": Js.Exn.t => unit,
18
+
[@bs.meth] "complete": unit => unit,
19
+
};
20
+
21
+
type observableT('a) = {
22
+
.
23
+
[@bs.meth] "subscribe": observerT('a) => subscriptionT,
24
+
};
25
+
26
+
type observableFactoryT('a) = (. unit) => observableT('a);
27
+
28
+
[@bs.get_index]
29
+
external observable_get:
30
+
(observableT('a), string) => option(observableFactoryT('a)) =
31
+
"";
32
+
[@bs.get_index]
33
+
external observable_unsafe_get:
34
+
(observableT('a), string) => observableFactoryT('a) =
35
+
"";
36
+
[@bs.set_index]
37
+
external observable_set:
38
+
(observableT('a), string, unit => observableT('a)) => unit =
39
+
"";
40
+
41
+
let fromObservable = (input: observableT('a)): sourceT('a) => {
42
+
let observable =
43
+
switch (input->observable_get(observableSymbol)) {
44
+
| Some(_) => (input->observable_unsafe_get(observableSymbol))(.)
45
+
| None => input
46
+
};
47
+
48
+
curry(sink => {
49
+
let observer: observerT('a) =
50
+
[@bs]
51
+
{
52
+
as _;
53
+
pub next = value => sink(. Push(value));
54
+
pub complete = () => sink(. End);
55
+
pub error = _ => ()
56
+
};
57
+
58
+
let subscription = observable##subscribe(observer);
59
+
60
+
sink(.
61
+
Start(
62
+
(. signal) =>
63
+
switch (signal) {
64
+
| Close => subscription##unsubscribe()
65
+
| _ => ()
66
+
},
67
+
),
68
+
);
69
+
});
70
+
};
71
+
72
+
type observableStateT = {
73
+
mutable talkback: (. talkbackT) => unit,
74
+
mutable ended: bool,
75
+
};
76
+
77
+
let toObservable = (source: sourceT('a)): observableT('a) => {
78
+
let observable: observableT('a) =
79
+
[@bs]
80
+
{
81
+
as _;
82
+
pub subscribe = (observer: observerT('a)): subscriptionT => {
83
+
let state: observableStateT = {
84
+
talkback: talkbackPlaceholder,
85
+
ended: false,
86
+
};
87
+
88
+
source((. signal) =>
89
+
switch (signal) {
90
+
| Start(x) =>
91
+
state.talkback = x;
92
+
x(. Pull);
93
+
| Push(x) when !state.ended =>
94
+
observer##next(x);
95
+
state.talkback(. Pull);
96
+
| Push(_) => ()
97
+
| End =>
98
+
state.ended = true;
99
+
observer##complete();
100
+
}
101
+
);
102
+
103
+
[@bs]
104
+
{
105
+
as _;
106
+
pub unsubscribe = () =>
107
+
if (!state.ended) {
108
+
state.ended = true;
109
+
state.talkback(. Close);
110
+
}
111
+
};
112
+
}
113
+
};
114
+
115
+
observable->observable_set(observableSymbol, () => observable);
116
+
observable;
117
+
};
+6
src/web/wonka_observable.rei
+6
src/web/wonka_observable.rei
+22
yarn.lock
+22
yarn.lock
···
2951
2951
resolved "https://registry.yarnpkg.com/call-me-maybe/-/call-me-maybe-1.0.1.tgz#26d208ea89e37b5cbde60250a15f031c16a4d66b"
2952
2952
integrity sha1-JtII6onje1y95gJQoV8DHBak1ms=
2953
2953
2954
+
callbag-from-iter@^1.2.0:
2955
+
version "1.2.0"
2956
+
resolved "https://registry.yarnpkg.com/callbag-from-iter/-/callbag-from-iter-1.2.0.tgz#c1886b08a447cd2efd9a140ec11743d705f26ca9"
2957
+
integrity sha512-9rWvHOnRGp01YMRHHwgVZOO1vu4IRR8GcoH3FpSB16AMzum5juFWJPCMX/XnkJ9j6cic/G+kvb1Grvi6IuSmIQ==
2958
+
2959
+
callbag-from-obs@^1.2.0:
2960
+
version "1.2.0"
2961
+
resolved "https://registry.yarnpkg.com/callbag-from-obs/-/callbag-from-obs-1.2.0.tgz#f092f302f302b53abaf1a4d7a5393f9a65fac517"
2962
+
integrity sha512-InhdPC6P4Gdpg7nuXSkocDFlb+//sbwCrVCYhxOHhSVm1gDcw/zSA+IF1gHdYtk4RQKKaCymUFCkVVUVSRThVQ==
2963
+
dependencies:
2964
+
symbol-observable "^1.2.0"
2965
+
2966
+
callbag-iterate@^1.0.0:
2967
+
version "1.0.0"
2968
+
resolved "https://registry.yarnpkg.com/callbag-iterate/-/callbag-iterate-1.0.0.tgz#97116f09296ef2d5073b35125891dca93349aeeb"
2969
+
integrity sha512-bynCbDuqGZkj1mXAhGr8jMf8Vhifps+G+pF3xlcz3jcaZLNXHghVjValnJtBTg2N74cyl347UzagJ8niJpyF6Q==
2970
+
2954
2971
caller-callsite@^2.0.0:
2955
2972
version "2.0.0"
2956
2973
resolved "https://registry.yarnpkg.com/caller-callsite/-/caller-callsite-2.0.0.tgz#847e0fce0a223750a9a027c54b33731ad3154134"
···
14931
14948
semver "^5.1.0"
14932
14949
strip-ansi "^5.0.0"
14933
14950
strip-bom "^3.0.0"
14951
+
14952
+
zen-observable@^0.8.14:
14953
+
version "0.8.14"
14954
+
resolved "https://registry.yarnpkg.com/zen-observable/-/zen-observable-0.8.14.tgz#d33058359d335bc0db1f0af66158b32872af3bf7"
14955
+
integrity sha512-kQz39uonEjEESwh+qCi83kcC3rZJGh4mrZW7xjkSQYXkq//JZHTtKo+6yuVloTgMtzsIWOJrjIrKvk/dqm0L5g==
14934
14956
14935
14957
zwitch@^1.0.0:
14936
14958
version "1.0.4"