Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Fix spec-compliance of toObservable operator (#94)
+35
-18
+35
-18
src/web/Wonka_observable.re
+35
-18
src/web/Wonka_observable.re
···
12
12
[@genType.import "../shims/Js.shim"]
13
13
type observableSubscriptionT = {. [@bs.meth] "unsubscribe": unit => unit};
14
14
15
+
[@bs.set_index]
16
+
external subscription_set: (observableSubscriptionT, string, bool) => unit;
17
+
15
18
[@genType.import "../shims/Js.shim"]
16
19
type observableObserverT('a) = {
17
20
.
···
30
33
31
34
[@bs.get_index]
32
35
external observable_get:
33
-
(observableT('a), string) => option(observableFactoryT('a)) =
34
-
"";
36
+
(observableT('a), string) => option(observableFactoryT('a));
35
37
[@bs.get_index]
36
38
external observable_unsafe_get:
37
-
(observableT('a), string) => observableFactoryT('a) =
38
-
"";
39
+
(observableT('a), string) => observableFactoryT('a);
39
40
[@bs.set_index]
40
41
external observable_set:
41
-
(observableT('a), string, unit => observableT('a)) => unit =
42
-
"";
42
+
(observableT('a), string, unit => observableT('a)) => unit;
43
43
44
44
[@genType]
45
45
let fromObservable = (input: observableT('a)): sourceT('a) => {
···
85
85
{
86
86
as _;
87
87
pub subscribe =
88
-
(observer: observableObserverT('a)): observableSubscriptionT => {
88
+
(_observer: observableObserverT('a)): observableSubscriptionT => {
89
+
let next: (. 'a) => unit = [%raw
90
+
{|
91
+
(typeof _observer === 'object' ? _observer.next : _observer) || function () {}
92
+
|}
93
+
];
94
+
95
+
let complete: (. unit) => unit = [%raw
96
+
{|
97
+
(typeof _observer === 'object' ? _observer.complete : arguments[2]) || function () {}
98
+
|}
99
+
];
100
+
89
101
let state: observableStateT = {
90
102
talkback: talkbackPlaceholder,
91
103
ended: false,
···
97
109
state.talkback = x;
98
110
x(. Pull);
99
111
| Push(x) when !state.ended =>
100
-
observer##next(x);
112
+
next(. x);
101
113
state.talkback(. Pull);
102
114
| Push(_) => ()
103
115
| End =>
104
116
state.ended = true;
105
-
observer##complete();
117
+
complete(.);
106
118
}
107
119
);
108
120
109
-
[@bs]
110
-
{
111
-
as _;
112
-
pub unsubscribe = () =>
113
-
if (!state.ended) {
114
-
state.ended = true;
115
-
state.talkback(. Close);
116
-
}
117
-
};
121
+
let subscription =
122
+
[@bs]
123
+
{
124
+
as self;
125
+
pub unsubscribe = () =>
126
+
if (!state.ended) {
127
+
self->subscription_set("closed", false);
128
+
state.ended = true;
129
+
state.talkback(. Close);
130
+
}
131
+
};
132
+
133
+
subscription->subscription_set("closed", false);
134
+
subscription;
118
135
}
119
136
};
120
137