kitten.sh / wonka
Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
0
fork

Configure Feed

Select the types of activity you want to include in your feed.

Fix combine for mixed cold/hot sources (#50)

When two sources are passed two combine of which
one is hot (actively sends Push signals without Pull
talkback signals) and one is cold (only sends Push
signals with Pull talkback signals) the combine
operator stalls.

It will receive a value from the hot source, but
wait for the cold source's value forever.

To prevent this, when no Pull talkback signal has
been received, one is sent to the other (presumably cold)
source proactively.

authored by

Phil Plückthun and committed by
GitHub b7c33ca0 592c41ea

+11 -3
+10 -2
src/wonka_operators.re
··· 101 101 | (Start(tb), _) => state.talkbackA = tb 102 102 | (Push(a), None) => 103 103 state.lastValA = Some(a); 104 - state.gotSignal = false; 104 + if (!state.gotSignal) { 105 + state.talkbackB(. Pull); 106 + } else { 107 + state.gotSignal = false; 108 + }; 105 109 | (Push(a), Some(b)) when !state.ended => 106 110 state.lastValA = Some(a); 107 111 state.gotSignal = false; ··· 120 124 | (Start(tb), _) => state.talkbackB = tb 121 125 | (Push(b), None) => 122 126 state.lastValB = Some(b); 123 - state.gotSignal = false; 127 + if (!state.gotSignal) { 128 + state.talkbackA(. Pull); 129 + } else { 130 + state.gotSignal = false; 131 + }; 124 132 | (Push(b), Some(a)) when !state.ended => 125 133 state.lastValB = Some(b); 126 134 state.gotSignal = false;
+1 -1
src/wonka_operators.test.ts
··· 313 313 const noop = (source: types.sourceT<any>) => operators.combine(sources.fromValue(0), source); 314 314 315 315 passesPassivePull(noop, [0, 0]); 316 - // TODO: passesActivePush(noop, [0, 0]); 316 + passesActivePush(noop, [0, 0]); 317 317 passesSinkClose(noop); 318 318 passesSourceEnd(noop, [0, 0]); 319 319 passesSingleStart(noop);