+24

src/callbag.re

reviewed

··· 122 122 | End => () 123 123 } 124 124 }); 125 125 + 126 126 + let subscribe = (f, source) => { 127 127 + let talkback = ref((_: talkbackT) => ()); 128 128 + let ended = ref(false); 129 129 + 130 130 + source(signal => { 131 131 + switch (signal) { 132 132 + | Start(x) => { 133 133 + talkback := x; 134 134 + talkback^(Pull); 135 135 + } 136 136 + | Push(x) when !ended^ => { 137 137 + f(x); 138 138 + talkback^(Pull); 139 139 + } 140 140 + | _ => () 141 141 + } 142 142 + }); 143 143 + 144 144 + () => if (!ended^) { 145 145 + ended := true; 146 146 + talkback^(End); 147 147 + } 148 148 + };

+1

src/callbag.rei

reviewed

··· 9 9 let merge: (array((signalT('a) => unit) => unit), signalT('a) => unit) => unit; 10 10 11 11 let forEach: ('a => unit, (signalT('a) => unit) => unit) => unit; 12 12 + let subscribe: ('a => unit, (signalT('a) => unit) => unit) => (unit => unit);