+34

__tests__/callbag_test.re

reviewed

··· 306 306 expect(res) |> toEqual([| (1, 2), (2, 2), (2, 4) |]); 307 307 }); 308 308 }); 309 309 + 310 310 + describe("take", () => { 311 311 + open Expect; 312 312 + 313 313 + it("only lets a maximum number of values through", () => { 314 314 + let talkback = ref((_: Callbag_types.talkbackT) => ()); 315 315 + let num = ref(1); 316 316 + 317 317 + let source = Callbag.take(2, sink => sink(Start(signal => { 318 318 + switch (signal) { 319 319 + | Pull => { 320 320 + let i = num^; 321 321 + num := num^ + 1; 322 322 + sink(Push(i)); 323 323 + } 324 324 + | _ => () 325 325 + } 326 326 + }))); 327 327 + 328 328 + let res = [||]; 329 329 + 330 330 + source(signal => { 331 331 + switch (signal) { 332 332 + | Start(x) => talkback := x 333 333 + | _ => ignore(Js.Array.push(signal, res)) 334 334 + } 335 335 + }); 336 336 + 337 337 + talkback^(Pull); 338 338 + talkback^(Pull); 339 339 + talkback^(Pull); 340 340 + expect(res) |> toEqual([| Push(1), Push(2), End |]); 341 341 + }); 342 342 + }); 309 343 }); 310 344 311 345 describe("sink factories", () => {

+27

src/callbag.re

reviewed

··· 237 237 })); 238 238 }; 239 239 240 240 + let take = (max, source, sink) => { 241 241 + let taken = ref(0); 242 242 + let talkback = ref((_: talkbackT) => ()); 243 243 + 244 244 + source(signal => { 245 245 + switch (signal) { 246 246 + | Start(tb) => { 247 247 + talkback := tb; 248 248 + sink(Start(signal => { 249 249 + if (taken^ < max) tb(signal); 250 250 + })); 251 251 + } 252 252 + | Push(_) when taken^ < max => { 253 253 + taken := taken^ + 1; 254 254 + sink(signal); 255 255 + 256 256 + if (taken^ === max) { 257 257 + sink(End); 258 258 + talkback^(End); 259 259 + }; 260 260 + } 261 261 + | End => sink(End) 262 262 + | _ => () 263 263 + } 264 264 + }); 265 265 + }; 266 266 + 240 267 let forEach = (f, source) => 241 268 captureTalkback(source, [@bs] (signal, talkback) => { 242 269 switch (signal) {

+6

src/callbag.rei

reviewed

··· 60 60 ) => 61 61 unit; 62 62 63 63 + /* Takes a max number and a source, and creates a sink & source. 64 64 + It will emit values that the sink receives until the passed maximum number 65 65 + of values is reached, at which point it will end the source and the 66 66 + returned, new source. */ 67 67 + let take: (int, (signalT('a) => unit) => unit, signalT('a) => unit) => unit; 68 68 + 63 69 /* -- sink factories */ 64 70 65 71 /* Takes a function and a source, and creates a sink.