Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Add combine operator with tests
+112
+42
__tests__/callbag_test.re
+42
__tests__/callbag_test.re
···
264
264
expect((numsA, nums)) |> toEqual(([| 1, 1 |], [| 1, 1, 2, 2 |]));
265
265
});
266
266
});
267
+
268
+
describe("combine", () => {
269
+
open Expect;
270
+
271
+
it("combines the latest values of two sources", () => {
272
+
let talkback = ref((_: Callbag_types.talkbackT) => ());
273
+
274
+
let makeSource = (factor: int) => {
275
+
let num = ref(1);
276
+
277
+
sink => {
278
+
sink(Start(signal => {
279
+
switch (signal) {
280
+
| Pull => {
281
+
let i = num^ * factor;
282
+
num := num^ + 1;
283
+
sink(Push(i));
284
+
}
285
+
| _ => ()
286
+
}
287
+
}));
288
+
}
289
+
};
290
+
291
+
let sourceA = makeSource(1);
292
+
let sourceB = makeSource(2);
293
+
let source = Callbag.combine(sourceA, sourceB);
294
+
let res = [||];
295
+
296
+
source(signal => {
297
+
switch (signal) {
298
+
| Start(x) => talkback := x
299
+
| Push(x) => ignore(Js.Array.push(x, res))
300
+
| _ => ()
301
+
}
302
+
});
303
+
304
+
talkback^(Pull);
305
+
talkback^(Pull);
306
+
expect(res) |> toEqual([| (1, 2), (2, 2), (2, 4) |]);
307
+
});
308
+
});
267
309
});
268
310
269
311
describe("sink factories", () => {
+69
src/callbag.re
+69
src/callbag.re
···
168
168
}
169
169
};
170
170
171
+
type combineLatestStateT('a, 'b) = {
172
+
mutable talkbackA: talkbackT => unit,
173
+
mutable talkbackB: talkbackT => unit,
174
+
mutable lastValA: option('a),
175
+
mutable lastValB: option('b),
176
+
mutable gotSignal: bool,
177
+
mutable endCounter: int,
178
+
mutable ended: bool,
179
+
};
180
+
181
+
let combine = (sourceA, sourceB, sink) => {
182
+
let state = {
183
+
talkbackA: (_: talkbackT) => (),
184
+
talkbackB: (_: talkbackT) => (),
185
+
lastValA: None,
186
+
lastValB: None,
187
+
gotSignal: false,
188
+
endCounter: 0,
189
+
ended: false
190
+
};
191
+
192
+
sourceA(signal => {
193
+
switch (signal, state.lastValB) {
194
+
| (Start(tb), _) => state.talkbackA = tb
195
+
| (Push(a), None) => state.lastValA = Some(a)
196
+
| (Push(a), Some(b)) when !state.ended => {
197
+
state.lastValA = Some(a);
198
+
state.gotSignal = false;
199
+
sink(Push((a, b)));
200
+
}
201
+
| (End, _) when state.endCounter < 2 => state.endCounter = state.endCounter + 1
202
+
| (End, _) => sink(End)
203
+
| _ => ()
204
+
}
205
+
});
206
+
207
+
sourceB(signal => {
208
+
switch (signal, state.lastValA) {
209
+
| (Start(tb), _) => state.talkbackB = tb
210
+
| (Push(b), None) => state.lastValB = Some(b)
211
+
| (Push(b), Some(a)) when !state.ended => {
212
+
state.lastValB = Some(b);
213
+
state.gotSignal = false;
214
+
sink(Push((a, b)));
215
+
}
216
+
| (End, _) when state.endCounter < 2 =>
217
+
state.endCounter = state.endCounter + 1
218
+
| (End, _) => sink(End)
219
+
| _ => ()
220
+
}
221
+
});
222
+
223
+
sink(Start(signal => {
224
+
switch (signal) {
225
+
| End => {
226
+
state.ended = true;
227
+
state.talkbackA(End);
228
+
state.talkbackB(End);
229
+
}
230
+
| Pull when !state.gotSignal => {
231
+
state.gotSignal = true;
232
+
state.talkbackA(signal);
233
+
state.talkbackB(signal);
234
+
}
235
+
| Pull => ()
236
+
}
237
+
}));
238
+
};
239
+
171
240
let forEach = (f, source) =>
172
241
captureTalkback(source, [@bs] (signal, talkback) => {
173
242
switch (signal) {
+1
src/callbag.rei
+1
src/callbag.rei
···
8
8
let scan: (('b, 'a) => 'b, 'b, (signalT('a) => unit) => unit, signalT('b) => unit) => unit;
9
9
let merge: (array((signalT('a) => unit) => unit), signalT('a) => unit) => unit;
10
10
let share: ((signalT('a) => unit) => unit, signalT('a) => unit) => unit;
11
+
let combine: ((signalT('a) => unit) => unit, (signalT('b) => unit) => unit, signalT(('a, 'b)) => unit) => unit;
11
12
12
13
let forEach: ('a => unit, (signalT('a) => unit) => unit) => unit;
13
14
let subscribe: ('a => unit, (signalT('a) => unit) => unit) => (unit => unit);