commit ee639330b0cfd1ed44851fe561bb5f646544af90 · kitten.sh/wonka

+22 -20

src/wonka_operators.re

···

       340
       340
        
       

     

       341
       341
        
               innerSource((. signal) =>

     

       342
       342
        
                 switch (signal) {

     

       343
       343
       -
                 | End =>

     

       344
       344
       -
                   state.innerTalkbacks =

     

       345
       345
       -
                     Rebel.Array.filter(state.innerTalkbacks, x => x !== talkback^);

     

       346
       346
       -
                   if (state.ended && Rebel.Array.size(state.innerTalkbacks) === 0) {

     

       347
       347
       -
                     sink(. End);

     

       348
       348
       -
                   };

     

       349
       343
        
                 | Start(tb) =>

     

       350
       344
        
                   talkback := tb;

     

       351
       345
        
                   state.innerTalkbacks =

     
···

       355
       349
        
                   sink(. Push(x));

     

       356
       350
        
                   talkback^(. Pull);

     

       357
       351
        
                 | Push(_) => ()

     

       352
       352
       +
                 | End when Rebel.Array.size(state.innerTalkbacks) !== 0 =>

     

       353
       353
       +
                   state.innerTalkbacks =

     

       354
       354
       +
                     Rebel.Array.filter(state.innerTalkbacks, x => x !== talkback^);

     

       355
       355
       +
                   if (state.ended && Rebel.Array.size(state.innerTalkbacks) === 0) {

     

       356
       356
       +
                     sink(. End);

     

       357
       357
       +
                   };

     

       358
       358
       +
                 | End => ()

     

       358
       359
        
                 }

     

       359
       360
        
               );

     

       360
       361
        
             };

     
···

       378
       379
        
             sink(.

     

       379
       380
        
               Start(

     

       380
       381
        
                 (. signal) =>

     

       381
       381
       -
                   switch (signal) {

     

       382
       382
       -
                   | Close when !state.ended =>

     

       383
       383
       -
                     let tbs = state.innerTalkbacks;

     

       384
       384
       -
                     state.innerTalkbacks = Rebel.Array.makeEmpty();

     

       385
       385
       -
                     state.outerTalkback(. signal);

     

       386
       386
       -
                     Rebel.Array.forEach(tbs, tb => tb(. signal));

     

       387
       387
       -
                   | Close => ()

     

       388
       388
       -
                   | Pull when !state.ended =>

     

       389
       389
       -
                     if (!state.outerPulled) {

     

       390
       390
       -
                       state.outerPulled = true;

     

       391
       391
       -
                       state.outerTalkback(. Pull);

     

       392
       392
       -
                     };

     

       382
       382
       +
                   if (!state.ended) {

     

       383
       383
       +
                     switch (signal) {

     

       384
       384
       +
                     | Close =>

     

       385
       385
       +
                       let tbs = state.innerTalkbacks;

     

       386
       386
       +
                       state.ended = true;

     

       387
       387
       +
                       state.innerTalkbacks = Rebel.Array.makeEmpty();

     

       388
       388
       +
                       state.outerTalkback(. signal);

     

       389
       389
       +
                       Rebel.Array.forEach(tbs, tb => tb(. signal));

     

       390
       390
       +
                     | Pull =>

     

       391
       391
       +
                       if (!state.outerPulled) {

     

       392
       392
       +
                         state.outerPulled = true;

     

       393
       393
       +
                         state.outerTalkback(. Pull);

     

       394
       394
       +
                       };

     

       393
       395
        
       

     

       394
       394
       -
                     Rebel.Array.forEach(state.innerTalkbacks, tb => tb(. Pull));

     

       395
       395
       -
                   | Pull => ()

     

       396
       396
       +
                       Rebel.Array.forEach(state.innerTalkbacks, tb => tb(. Pull));

     

       397
       397
       +
                     };

     

       396
       398
        
                   },

     

       397
       399
        
               ),

     

       398
       400
        
             );

+28 -2

src/wonka_operators.test.ts

···

       201
       201
        
         This isn't a strict requirement, but some operators should ensure that

     

       202
       202
        
         all sources are well behaved. This is particularly true for operators

     

       203
       203
        
         that either Close sources themselves or may operate on multiple sources. */

     

       204
       204
       -
       const passesStrictEnd = (operator: types.operatorT<any, any>) =>

     

       204
       204
       +
       const passesStrictEnd = (operator: types.operatorT<any, any>) => {

     

       205
       205
        
         it('stops all signals after End has been received (spec: strict end)', () => {

     

       206
       206
        
           let pulls = 0;

     

       207
       207
        
           const signals = [];

     
···

       231
       231
        
           expect(signals).toEqual([deriving.end()]);

     

       232
       232
        
           expect(pulls).toBe(1);

     

       233
       233
        
         });

     

       234
       234
       +
       

     

       235
       235
       +
         it('stops all signals after Close has been received (spec: strict close)', () => {

     

       236
       236
       +
           const signals = [];

     

       237
       237
       +
       

     

       238
       238
       +
           const source: types.sourceT<any> = sink => {

     

       239
       239
       +
             sink(deriving.start(tb => {

     

       240
       240
       +
               if (tb === deriving.close) {

     

       241
       241
       +
                 sink(deriving.push(123));

     

       242
       242
       +
               }

     

       243
       243
       +
             }));

     

       244
       244
       +
           };

     

       245
       245
       +
       

     

       246
       246
       +
           const sink: types.sinkT<any> = signal => {

     

       247
       247
       +
             if (deriving.isStart(signal)) {

     

       248
       248
       +
               deriving.unboxStart(signal)(deriving.close);

     

       249
       249
       +
             } else {

     

       250
       250
       +
               signals.push(signal);

     

       251
       251
       +
             }

     

       252
       252
       +
           };

     

       253
       253
       +
       

     

       254
       254
       +
           operator(source)(sink);

     

       255
       255
       +
       

     

       256
       256
       +
           // The Push signal should've been dropped

     

       257
       257
       +
           jest.runAllTimers();

     

       258
       258
       +
           expect(signals).toEqual([]);

     

       259
       259
       +
         });

     

       260
       260
       +
       };

     

       234
       261
        
       

     

       235
       262
        
       /* This tests an immediately closing operator for End signals to

     

       236
       263
        
         the sink and Close signals to the source.

     
···

       968
       995
        
         passesSinkClose(noop);

     

       969
       996
        
         passesSourceEnd(noop);

     

       970
       997
        
         passesSingleStart(noop);

     

       971
       971
       -
         passesStrictEnd(noop);

     

       972
       998
        
         passesAsyncSequence(noop);

     

       973
       999
        
       

     

       974
       1000
        
         const ending = operators.takeWhile(() => false);