Mirror: 🎩 A tiny but capable push & pull stream library for TypeScript and Flow
Add strict ending to onEnd and onPush (#61)
Since these are side-effects that may be used
to reinsert more signals into the source chain,
through subjects for instance, they need to
protect themselves from dangerous signal loops.
authored by
Phil Plückthun
and committed by
GitHub
00c9ac25
1439d0b8
+46
-18
+44
-18
src/wonka_operators.re
+44
-18
src/wonka_operators.re
···
434
434
let onEnd = (f: (. unit) => unit): operatorT('a, 'a) =>
435
435
curry(source =>
436
436
curry(sink => {
437
+
let ended = ref(false);
437
438
source((. signal) =>
438
439
switch (signal) {
439
440
| Start(talkback) =>
440
441
sink(.
441
442
Start(
442
-
(. signal) => {
443
-
switch (signal) {
444
-
| Close => f(.)
445
-
| _ => ()
446
-
};
447
-
talkback(. signal);
448
-
},
443
+
(. signal) =>
444
+
if (! ended^) {
445
+
switch (signal) {
446
+
| Pull => talkback(. signal)
447
+
| Close =>
448
+
ended := true;
449
+
talkback(. signal);
450
+
f(.);
451
+
};
452
+
},
449
453
),
450
454
)
451
-
| End =>
455
+
| Push(_) when ! ended^ => sink(. signal)
456
+
| Push(_) => ()
457
+
| End when ! ended^ =>
458
+
ended := true;
452
459
sink(. signal);
453
460
f(.);
454
-
| _ => sink(. signal)
461
+
| End => ()
455
462
}
456
-
)
463
+
);
457
464
})
458
465
);
459
466
460
467
[@genType]
461
468
let onPush = (f: (. 'a) => unit): operatorT('a, 'a) =>
462
469
curry(source =>
463
-
curry(sink =>
470
+
curry(sink => {
471
+
let ended = ref(false);
464
472
source((. signal) => {
465
473
switch (signal) {
466
-
| Push(x) => f(. x)
467
-
| _ => ()
468
-
};
469
-
470
-
sink(. signal);
471
-
})
472
-
)
474
+
| Start(talkback) =>
475
+
sink(.
476
+
Start(
477
+
(. signal) =>
478
+
if (! ended^) {
479
+
switch (signal) {
480
+
| Pull => talkback(. signal)
481
+
| Close =>
482
+
ended := true;
483
+
talkback(. signal);
484
+
};
485
+
},
486
+
),
487
+
)
488
+
| Push(x) when ! ended^ =>
489
+
f(. x);
490
+
sink(. signal);
491
+
| Push(_) => ()
492
+
| End when ! ended^ =>
493
+
ended := true;
494
+
sink(. signal);
495
+
| End => ()
496
+
}
497
+
});
498
+
})
473
499
);
474
500
475
501
[@genType]
+2
src/wonka_operators.test.ts
+2
src/wonka_operators.test.ts
···
758
758
passesActivePush(noop);
759
759
passesSinkClose(noop);
760
760
passesSourceEnd(noop);
761
+
passesStrictEnd(noop);
761
762
passesSingleStart(noop);
762
763
passesAsyncSequence(noop);
763
764
···
781
782
passesActivePush(noop);
782
783
passesSinkClose(noop);
783
784
passesSourceEnd(noop);
785
+
passesStrictEnd(noop);
784
786
passesSingleStart(noop);
785
787
passesAsyncSequence(noop);
786
788