commit d0ee5131d6db0332a9b2385be10c97999848f824 · kitten.sh/wonka

+5 -14

src/wonka_operators.re

···

       412
        
       let onEnd = (f: (. unit) => unit): operatorT('a, 'a) =>

     

       413
        
         curry(source =>

     

       414
        
           curry(sink => {

     

       415
       -
             let ended = ref(false);

     

       416
       -
       

     

       417
        
             source((. signal) =>

     

       418
        
               switch (signal) {

     

       419
        
               | Start(talkback) =>

     
···

       421
        
                   Start(

     

       422
        
                     (. signal) => {

     

       423
        
                       switch (signal) {

     

       424
       -
                       | Close when ! ended^ =>

     

       425
       -
                         ended := true;

     

       426
       -
                         f(.);

     

       427
       -
                       | Close

     

       428
       -
                       | Pull => ()

     

       429
        
                       };

     

       430
       -
       

     

       431
        
                       talkback(. signal);

     

       432
        
                     },

     

       433
        
                   ),

     

       434
        
                 )

     

       435
        
               | End =>

     

       436
       -
                 if (! ended^) {

     

       437
       -
                   ended := true;

     

       438
       -
                   sink(. signal);

     

       439
       -
                   f(.);

     

       440
       -
                 }

     

       441
        
               | _ => sink(. signal)

     

       442
        
               }

     

       443
       -
             );

     

       444
        
           })

     

       445
        
         );

     

       446

···

       412
        
       let onEnd = (f: (. unit) => unit): operatorT('a, 'a) =>

     

       413
        
         curry(source =>

     

       414
        
           curry(sink => {

     

       0
        
       
     

       0
        
       
     

       415
        
             source((. signal) =>

     

       416
        
               switch (signal) {

     

       417
        
               | Start(talkback) =>

     
···

       419
        
                   Start(

     

       420
        
                     (. signal) => {

     

       421
        
                       switch (signal) {

     

       422
       +
                       | Close => f(.)

     

       423
       +
                       | _ => ()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       424
        
                       };

     

       0
        
       
     

       425
        
                       talkback(. signal);

     

       426
        
                     },

     

       427
        
                   ),

     

       428
        
                 )

     

       429
        
               | End =>

     

       430
       +
                 sink(. signal);

     

       431
       +
                 f(.);

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       432
        
               | _ => sink(. signal)

     

       433
        
               }

     

       434
       +
             )

     

       435
        
           })

     

       436
        
         );

     

       437

-1

src/wonka_operators.test.ts

···

       538
        
         passesSinkClose(noop);

     

       539
        
         passesSourceEnd(noop);

     

       540
        
         passesSingleStart(noop);

     

       541
       -
         // TODO: passesStrictEnd(noop);

     

       542
        
         passesAsyncSequence(noop);

     

       543
        
       

     

       544
        
         it('maps over values given a transform function', () => {

···

       538
        
         passesSinkClose(noop);

     

       539
        
         passesSourceEnd(noop);

     

       540
        
         passesSingleStart(noop);

     

       0
        
       
     

       541
        
         passesAsyncSequence(noop);

     

       542
        
       

     

       543
        
         it('maps over values given a transform function', () => {

+40 -11

src/wonka_sources.re

···

       46
        
                   ended := true;

     

       47
        
                   sink(. Push(x));

     

       48
        
                   sink(. End);

     

       49
       -
                 | _ => ()

     

       0
        
       
     

       50
        
                 },

     

       51
        
             ),

     

       52
        
           );

     

       53
        
         });

     

       54
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       55
        
       [@genType]

     

       56
        
       let make = (f: (. observerT('a)) => teardownT): sourceT('a) =>

     

       57
        
         curry(sink => {

     

       58
       -
           let teardown = ref((.) => ());

     

       59
        
       

     

       60
        
           sink(.

     

       61
        
             Start(

     

       62
        
               (. signal) =>

     

       63
        
                 switch (signal) {

     

       64
       -
                 | Close => teardown^(.)

     

       65
       -
                 | Pull => ()

     

       0
        
       
     

       0
        
       
     

       66
        
                 },

     

       67
        
             ),

     

       68
        
           );

     

       69
        
       

     

       70
       -
           teardown :=

     

       71
        
             f(. {

     

       72
       -
               next: value => sink(. Push(value)),

     

       73
       -
               complete: () => sink(. End),

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       74
        
             });

     

       75
        
         });

     

       76
        
       

     
···

       91
        
           sink(.

     

       92
        
             Start(

     

       93
        
               (. signal) =>

     

       94
       -
                 if (signal === Close) {

     

       95
       -
                   state.sinks = Rebel.Array.filter(state.sinks, x => x !== sink);

     

       0
        
       
     

       0
        
       
     

       96
        
                 },

     

       97
        
             ),

     

       98
        
           );

     
···

       114
        
       

     

       115
        
       [@genType]

     

       116
        
       let empty = (sink: sinkT('a)): unit => {

     

       117
       -
         sink(. Start(talkbackPlaceholder));

     

       118
       -
         sink(. End);

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       119
        
       };

     

       120
        
       

     

       121
        
       [@genType]

···

       46
        
                   ended := true;

     

       47
        
                   sink(. Push(x));

     

       48
        
                   sink(. End);

     

       49
       +
                 | Pull => ()

     

       50
       +
                 | Close => ended := true

     

       51
        
                 },

     

       52
        
             ),

     

       53
        
           );

     

       54
        
         });

     

       55
        
       

     

       56
       +
       type makeStateT = {

     

       57
       +
         mutable teardown: (. unit) => unit,

     

       58
       +
         mutable ended: bool,

     

       59
       +
       };

     

       60
       +
       

     

       61
        
       [@genType]

     

       62
        
       let make = (f: (. observerT('a)) => teardownT): sourceT('a) =>

     

       63
        
         curry(sink => {

     

       64
       +
           let state: makeStateT = {teardown: (.) => (), ended: false};

     

       65
        
       

     

       66
        
           sink(.

     

       67
        
             Start(

     

       68
        
               (. signal) =>

     

       69
        
                 switch (signal) {

     

       70
       +
                 | Close when !state.ended =>

     

       71
       +
                   state.ended = true;

     

       72
       +
                   state.teardown(.);

     

       73
       +
                 | _ => ()

     

       74
        
                 },

     

       75
        
             ),

     

       76
        
           );

     

       77
        
       

     

       78
       +
           state.teardown =

     

       79
        
             f(. {

     

       80
       +
               next: value =>

     

       81
       +
                 if (!state.ended) {

     

       82
       +
                   sink(. Push(value));

     

       83
       +
                 },

     

       84
       +
               complete: () =>

     

       85
       +
                 if (!state.ended) {

     

       86
       +
                   state.ended = true;

     

       87
       +
                   sink(. End);

     

       88
       +
                 },

     

       89
        
             });

     

       90
        
         });

     

       91
        
       

     
···

       106
        
           sink(.

     

       107
        
             Start(

     

       108
        
               (. signal) =>

     

       109
       +
                 switch (signal) {

     

       110
       +
                 | Close =>

     

       111
       +
                   state.sinks = Rebel.Array.filter(state.sinks, x => x !== sink)

     

       112
       +
                 | _ => ()

     

       113
        
                 },

     

       114
        
             ),

     

       115
        
           );

     
···

       131
        
       

     

       132
        
       [@genType]

     

       133
        
       let empty = (sink: sinkT('a)): unit => {

     

       134
       +
         let ended = ref(false);

     

       135
       +
         sink(.

     

       136
       +
           Start(

     

       137
       +
             (. signal) => {

     

       138
       +
               switch (signal) {

     

       139
       +
               | Close => ended := true

     

       140
       +
               | _ => ()

     

       141
       +
               }

     

       142
       +
             },

     

       143
       +
           ),

     

       144
       +
         );

     

       145
       +
         if (! ended^) {

     

       146
       +
           sink(. End);

     

       147
       +
         };

     

       148
        
       };

     

       149
        
       

     

       150
        
       [@genType]