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 close talkback on {concat,merge,switch}Map

The Close event was not being respected and sent
to the inner talkback when the outer source has
already ended.

+21 -16
+7 -5
src/operators/wonka_operator_concatMap.re
··· 75 75 sink(.Start((.signal) => { 76 76 switch (signal) { 77 77 | Pull => if (!state.ended) state.innerTalkback(.Pull) 78 - | Close when !state.ended => { 79 - state.ended = true; 80 - state.closed = true; 81 - state.outerTalkback(.Close); 78 + | Close => { 82 79 state.innerTalkback(.Close); 80 + if (!state.ended) { 81 + state.ended = true; 82 + state.closed = true; 83 + state.outerTalkback(.Close); 84 + state.innerTalkback = talkbackPlaceholder; 85 + } 83 86 } 84 - | Close => () 85 87 } 86 88 })); 87 89 }));
+7 -5
src/operators/wonka_operator_mergeMap.re
··· 62 62 63 63 sink(.Start((.signal) => { 64 64 switch (signal) { 65 - | Close when !state.ended => { 66 - state.ended = true; 67 - state.outerTalkback(.Close); 65 + | Close => { 68 66 Rebel.Array.forEach(state.innerTalkbacks, talkback => talkback(.Close)); 69 - state.innerTalkbacks = Rebel.Array.makeEmpty(); 67 + if (!state.ended) { 68 + state.ended = true; 69 + state.outerTalkback(.Close); 70 + Rebel.Array.forEach(state.innerTalkbacks, talkback => talkback(.Close)); 71 + state.innerTalkbacks = Rebel.Array.makeEmpty(); 72 + } 70 73 } 71 - | Close => () 72 74 | Pull when !state.ended => 73 75 Rebel.Array.forEach(state.innerTalkbacks, talkback => talkback(.Pull)); 74 76 | Pull => ()
+7 -6
src/operators/wonka_operator_switchMap.re
··· 65 65 sink(.Start((.signal) => { 66 66 switch (signal) { 67 67 | Pull => state.innerTalkback(.Pull) 68 - | Close when !state.ended => { 69 - state.ended = true; 70 - state.closed = true; 71 - state.outerTalkback(.Close); 68 + | Close => { 72 69 state.innerTalkback(.Close); 73 - state.innerTalkback = talkbackPlaceholder; 70 + if (!state.ended) { 71 + state.ended = true; 72 + state.closed = true; 73 + state.outerTalkback(.Close); 74 + state.innerTalkback = talkbackPlaceholder; 75 + } 74 76 } 75 - | Close => () 76 77 } 77 78 })); 78 79 }));