Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Return unit (undefined) where possible
+54
-39
+50
-36
src/Wonka_operators.re
+50
-36
src/Wonka_operators.re
···
21
21
ended: false,
22
22
};
23
23
24
-
source((. signal) =>
24
+
source((. signal) => {
25
25
switch (signal) {
26
26
| Start(tb) =>
27
27
state.sourceTalkback = tb;
28
28
29
-
notifier((. signal) =>
29
+
notifier((. signal) => {
30
30
switch (signal) {
31
31
| Start(tb) => state.notifierTalkback = tb
32
32
| Push(_) when !state.ended =>
···
44
44
};
45
45
sink(. End);
46
46
| End => ()
47
-
}
48
-
);
47
+
};
48
+
();
49
+
});
49
50
| Push(value) when !state.ended =>
50
51
Rebel.MutableQueue.add(state.buffer, value);
51
52
if (!state.pulled) {
···
64
65
};
65
66
sink(. End);
66
67
| End => ()
67
-
}
68
-
);
68
+
};
69
+
();
70
+
});
69
71
70
72
sink(.
71
73
Start(
···
204
206
205
207
let rec applyInnerSource = innerSource => {
206
208
state.innerActive = true;
207
-
innerSource((. signal) =>
209
+
innerSource((. signal) => {
208
210
switch (signal) {
209
211
| Start(tb) =>
210
212
state.innerTalkback = tb;
···
229
231
| None => ()
230
232
};
231
233
| End => ()
232
-
}
233
-
);
234
+
};
235
+
();
236
+
});
237
+
();
234
238
};
235
239
236
-
source((. signal) =>
240
+
source((. signal) => {
237
241
switch (signal) {
238
242
| Start(tb) => state.outerTalkback = tb
239
243
| Push(x) when !state.ended =>
···
251
255
sink(. End);
252
256
};
253
257
| End => ()
254
-
}
255
-
);
258
+
};
259
+
();
260
+
});
256
261
257
262
sink(.
258
263
Start(
···
296
301
curry(sink => {
297
302
let talkback = ref(talkbackPlaceholder);
298
303
299
-
source((. signal) =>
304
+
source((. signal) => {
300
305
switch (signal) {
301
306
| Start(tb) =>
302
307
talkback := tb;
303
308
sink(. signal);
304
309
| Push(x) when !f(. x) => talkback^(. Pull)
305
310
| _ => sink(. signal)
306
-
}
307
-
);
311
+
};
312
+
();
313
+
});
308
314
})
309
315
);
310
316
···
312
318
let map = (f: (. 'a) => 'b): operatorT('a, 'b) =>
313
319
curry(source =>
314
320
curry(sink =>
315
-
source((. signal) =>
321
+
source((. signal) => {
316
322
sink(.
317
323
/* The signal needs to be recreated for genType to generate
318
324
the correct generics during codegen */
···
322
328
| End => End
323
329
},
324
330
)
325
-
)
331
+
})
326
332
)
327
333
);
328
334
···
494
500
ended := true;
495
501
sink(. signal);
496
502
| End => ()
497
-
}
503
+
};
504
+
();
498
505
});
499
506
})
500
507
);
···
708
715
ended: false,
709
716
};
710
717
711
-
source((. signal) =>
718
+
source((. signal) => {
712
719
switch (signal) {
713
720
| Start(tb) =>
714
721
state.sourceTalkback = tb;
715
722
716
-
notifier((. signal) =>
723
+
notifier((. signal) => {
717
724
switch (signal) {
718
725
| Start(innerTb) =>
719
726
state.notifierTalkback = innerTb;
···
725
732
state.ended = true;
726
733
state.sourceTalkback(. Close);
727
734
| End => ()
728
-
}
729
-
);
735
+
};
736
+
();
737
+
});
730
738
| Push(_) when !state.skip && !state.ended =>
731
739
state.pulled = false;
732
740
sink(. signal);
···
741
749
};
742
750
state.ended = true;
743
751
sink(. End);
744
-
}
745
-
);
752
+
};
753
+
();
754
+
});
746
755
747
756
sink(.
748
757
Start(
···
850
859
};
851
860
}
852
861
);
862
+
();
853
863
};
854
864
855
-
source((. signal) =>
865
+
source((. signal) => {
856
866
switch (signal) {
857
867
| Start(tb) => state.outerTalkback = tb
858
868
| Push(x) when !state.ended =>
···
876
886
sink(. End);
877
887
};
878
888
| End => ()
879
-
}
880
-
);
889
+
};
890
+
();
891
+
});
881
892
882
893
sink(.
883
894
Start(
···
981
992
talkback: talkbackPlaceholder,
982
993
};
983
994
984
-
source((. signal) =>
995
+
source((. signal) => {
985
996
switch (signal) {
986
997
| Start(talkback) when max <= 0 =>
987
998
talkback(. Close);
···
1002
1013
Rebel.MutableQueue.toArray(state.queue),
1003
1014
sink,
1004
1015
)
1005
-
}
1006
-
);
1016
+
};
1017
+
();
1018
+
});
1007
1019
})
1008
1020
);
1009
1021
···
1023
1035
notifierTalkback: talkbackPlaceholder,
1024
1036
};
1025
1037
1026
-
source((. signal) =>
1038
+
source((. signal) => {
1027
1039
switch (signal) {
1028
1040
| Start(tb) =>
1029
1041
state.sourceTalkback = tb;
1030
1042
1031
-
notifier((. signal) =>
1043
+
notifier((. signal) => {
1032
1044
switch (signal) {
1033
1045
| Start(innerTb) =>
1034
1046
state.notifierTalkback = innerTb;
···
1038
1050
state.sourceTalkback(. Close);
1039
1051
sink(. End);
1040
1052
| End => ()
1041
-
}
1042
-
);
1053
+
};
1054
+
();
1055
+
});
1043
1056
| End when !state.ended =>
1044
1057
state.ended = true;
1045
1058
state.notifierTalkback(. Close);
···
1047
1060
| End => ()
1048
1061
| Push(_) when !state.ended => sink(. signal)
1049
1062
| Push(_) => ()
1050
-
}
1051
-
);
1063
+
};
1064
+
();
1065
+
});
1052
1066
1053
1067
sink(.
1054
1068
Start(
+4
-3
src/web/WonkaJs.re
+4
-3
src/web/WonkaJs.re
···
160
160
/* sinks */
161
161
[@genType]
162
162
let toPromise = (source: sourceT('a)): Js.Promise.t('a) => {
163
-
Js.Promise.make((~resolve, ~reject as _) =>
163
+
Js.Promise.make((~resolve, ~reject as _) => {
164
164
Wonka_operators.takeLast(1, source, (. signal) =>
165
165
switch (signal) {
166
166
| Start(x) => x(. Pull)
167
167
| Push(x) => resolve(. x)
168
168
| End => ()
169
169
}
170
-
)
171
-
);
170
+
);
171
+
();
172
+
});
172
173
};
173
174
174
175
/* sources */