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.

Add missing Close signal to takeUntil operator (#128)

authored by

Phil Pluckthun and committed by
GitHub
2578f76c cc88647f

+53
+52
src/__tests__/operators.test.ts
··· 506 506 expect(fnB).toHaveBeenCalledWith([0]); 507 507 expect(fnA.mock.calls[0][0]).toBe(fnB.mock.calls[0][0]); 508 508 }); 509 + 510 + it('completes the source when no more sink is listening', () => { 511 + let onPush = () => {}; 512 + 513 + const talkback = vi.fn(); 514 + const source: Source<any> = operators.share(sink => { 515 + sink(start(talkback)); 516 + onPush = () => { 517 + sink(push([0])); 518 + sink(push([1])); 519 + sink(SignalKind.End); 520 + }; 521 + }); 522 + 523 + const fnA = vi.fn(); 524 + const fnB = vi.fn(); 525 + 526 + sinks.forEach(fnA)(operators.take(1)(source)); 527 + sinks.forEach(fnB)(operators.take(1)(source)); 528 + onPush(); 529 + 530 + expect(fnA).toHaveBeenCalledWith([0]); 531 + expect(fnB).toHaveBeenCalledWith([0]); 532 + expect(fnA.mock.calls[0][0]).toBe(fnB.mock.calls[0][0]); 533 + expect(talkback).toHaveBeenCalledWith(TalkbackKind.Close); 534 + }); 509 535 }); 510 536 511 537 describe('skip', () => { ··· 696 722 notify(null); 697 723 expect(fn).toHaveBeenCalledTimes(3); 698 724 expect(fn.mock.calls[2][0]).toEqual(SignalKind.End); 725 + }); 726 + 727 + it('emits values until a notifier emits', () => { 728 + const { source: input$, next } = sources.makeSubject<number>(); 729 + const fn = vi.fn(); 730 + 731 + let hasClosed = false; 732 + 733 + operators.takeUntil(sink => { 734 + sink( 735 + start(talkback => { 736 + if (talkback === TalkbackKind.Close) { 737 + hasClosed = true; 738 + } else if (talkback === TalkbackKind.Pull && !hasClosed) { 739 + sink(push(1)); 740 + } 741 + }) 742 + ); 743 + })(input$)(fn); 744 + 745 + next(1); 746 + 747 + expect(fn).toHaveBeenCalledTimes(2); 748 + expect(fn.mock.calls).toEqual([[0], [start(expect.any(Function))]]); 749 + 750 + expect(hasClosed).toBe(true); 699 751 }); 700 752 }); 701 753
+1
src/operators.ts
··· 710 710 (notifierTalkback = signal[0])(TalkbackKind.Pull); 711 711 } else { 712 712 ended = true; 713 + notifierTalkback(TalkbackKind.Close); 713 714 sourceTalkback(TalkbackKind.Close); 714 715 sink(SignalKind.End); 715 716 }