···

       
162
       
 
       
      let nums = [||];

     

       
163
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
164
       
 
       


     

       
165
       
-
       
      Wonka.map((_) => {

     

       
166
       
 
       
        let res = num^;

     

       
167
       
 
       
        num := num^ + 1;

     

       
168
       
 
       
        res

     
···

       
191
       
 
       
    });

     

       
192
       
 
       


     

       
193
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
194
       
-
       
      Wonka_thelpers.testWithListenable(Wonka.map(x => x))

     

       
195
       
 
       
        |> Js.Promise.then_(x => {

     

       
196
       
 
       
          expect(x)

     

       
197
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
202
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
203
       
 
       
      let end_: talkbackT = Close;

     

       
204
       
 
       


     

       
205
       
-
       
      Wonka_thelpers.testTalkbackEnd(Wonka.map(x => x))

     

       
206
       
 
       
        |> Js.Promise.then_(x => {

     

       
207
       
 
       
          expect(x)

     

       
208
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
219
       
 
       
      let nums = [||];

     

       
220
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
221
       
 
       


     

       
222
       
-
       
      Wonka.filter(x => x mod 2 === 0, sink => {

     

       
223
       
 
       
        sink(.Start((.signal) => {

     

       
224
       
 
       
          switch (signal) {

     

       
225
       
 
       
          | Pull => {

     
···

       
248
       
 
       
    });

     

       
249
       
 
       


     

       
250
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
251
       
-
       
      Wonka_thelpers.testWithListenable(Wonka.filter((_) => true))

     

       
252
       
 
       
        |> Js.Promise.then_(x => {

     

       
253
       
 
       
          expect(x)

     

       
254
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
257
       
 
       
    });

     

       
258
       
 
       


     

       
259
       
 
       
    testPromise("follows the spec for listenables when filtering", () => {

     

       
260
       
-
       
      Wonka_thelpers.testWithListenable(Wonka.filter((_) => false))

     

       
261
       
 
       
        |> Js.Promise.then_(x => {

     

       
262
       
 
       
          expect(x)

     

       
263
       
 
       
            |> toEqual(([| Pull, Pull |], [| End |]))

     
···

       
268
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
269
       
 
       
      let end_: talkbackT = Close;

     

       
270
       
 
       


     

       
271
       
-
       
      Wonka_thelpers.testTalkbackEnd(Wonka.filter((_) => true))

     

       
272
       
 
       
        |> Js.Promise.then_(x => {

     

       
273
       
 
       
          expect(x)

     

       
274
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
742
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
743
       
 
       
      let num = ref(1);

     

       
744
       
 
       


     

       
745
       
-
       
      let source = Wonka.takeWhile(x => x <= 2, sink => sink(.Start((.signal) => {

     

       
746
       
 
       
        switch (signal) {

     

       
747
       
 
       
        | Pull => {

     

       
748
       
 
       
          let i = num^;

     
···

       
774
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
775
       
 
       
      let num = ref(1);

     

       
776
       
 
       


     

       
777
       
-
       
      let source = Wonka.takeWhile(x => x <= 5, sink => sink(.Start((.signal) => {

     

       
778
       
 
       
        switch (signal) {

     

       
779
       
 
       
        | Pull => {

     

       
780
       
 
       
          let i = num^;

     
···

       
806
       
 
       
    });

     

       
807
       
 
       


     

       
808
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
809
       
-
       
      Wonka_thelpers.testWithListenable(Wonka.takeWhile((_) => true))

     

       
810
       
 
       
        |> Js.Promise.then_(x => {

     

       
811
       
 
       
          expect(x)

     

       
812
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
817
       
 
       
    testPromise("follows the spec for listenables when ending the source", () => {

     

       
818
       
 
       
      let end_: talkbackT = Close;

     

       
819
       
 
       


     

       
820
       
-
       
      Wonka_thelpers.testWithListenable(Wonka.takeWhile((_) => false))

     

       
821
       
 
       
        |> Js.Promise.then_(x => {

     

       
822
       
 
       
          expect(x)

     

       
823
       
 
       
            |> toEqual(([| end_ |], [| End |]))

     
···

       
828
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
829
       
 
       
      let end_: talkbackT = Close;

     

       
830
       
 
       


     

       
831
       
-
       
      Wonka_thelpers.testTalkbackEnd(Wonka.takeWhile((_) => true))

     

       
832
       
 
       
        |> Js.Promise.then_(x => {

     

       
833
       
 
       
          expect(x)

     

       
834
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
1023
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
1024
       
 
       
      let num = ref(1);

     

       
1025
       
 
       


     

       
1026
       
-
       
      let source = Wonka.skipWhile(x => x <= 2, sink => sink(.Start((.signal) => {

     

       
1027
       
 
       
        switch (signal) {

     

       
1028
       
 
       
        | Pull when num^ <= 4 => {

     

       
1029
       
 
       
          let i = num^;

     
···

       
1051
       
 
       
    });

     

       
1052
       
 
       


     

       
1053
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
1054
       
-
       
      Wonka_thelpers.testWithListenable(Wonka.skipWhile((_) => false))

     

       
1055
       
 
       
        |> Js.Promise.then_(x => {

     

       
1056
       
 
       
          expect(x)

     

       
1057
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
1060
       
 
       
    });

     

       
1061
       
 
       


     

       
1062
       
 
       
    testPromise("follows the spec for listenables when skipping the source", () => {

     

       
1063
       
-
       
      Wonka_thelpers.testWithListenable(Wonka.skipWhile((_) => true))

     

       
1064
       
 
       
        |> Js.Promise.then_(x => {

     

       
1065
       
 
       
          expect(x)

     

       
1066
       
 
       
            |> toEqual(([| Pull, Pull |], [| End |]))

     
···

       
1071
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
1072
       
 
       
      let end_: talkbackT = Close;

     

       
1073
       
 
       


     

       
1074
       
-
       
      Wonka_thelpers.testTalkbackEnd(Wonka.skipWhile((_) => false))

     

       
1075
       
 
       
        |> Js.Promise.then_(x => {

     

       
1076
       
 
       
          expect(x)

     

       
1077
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
1288
       
 
       


     

       
1289
       
 
       
    input

     

       
1290
       
 
       
      |> Wonka.fromArray

     

       
1291
       
-
       
      |> Wonka.map(x => string_of_int(x))

     

       
1292
       
 
       
      |> Wonka.forEach((.x) => ignore(Js.Array.push(x, actual)));

     

       
1293
       
 
       


     

       
1294
       
 
       
    expect(output) |> toEqual(output)

     

···

       
162
       
 
       
      let nums = [||];

     

       
163
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
164
       
 
       


     

       
165
       
+
       
      Wonka.map((._) => {

     

       
166
       
 
       
        let res = num^;

     

       
167
       
 
       
        num := num^ + 1;

     

       
168
       
 
       
        res

     
···

       
191
       
 
       
    });

     

       
192
       
 
       


     

       
193
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
194
       
+
       
      Wonka_thelpers.testWithListenable(Wonka.map((.x) => x))

     

       
195
       
 
       
        |> Js.Promise.then_(x => {

     

       
196
       
 
       
          expect(x)

     

       
197
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
202
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
203
       
 
       
      let end_: talkbackT = Close;

     

       
204
       
 
       


     

       
205
       
+
       
      Wonka_thelpers.testTalkbackEnd(Wonka.map((.x) => x))

     

       
206
       
 
       
        |> Js.Promise.then_(x => {

     

       
207
       
 
       
          expect(x)

     

       
208
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
219
       
 
       
      let nums = [||];

     

       
220
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
221
       
 
       


     

       
222
       
+
       
      Wonka.filter((.x) => x mod 2 === 0, sink => {

     

       
223
       
 
       
        sink(.Start((.signal) => {

     

       
224
       
 
       
          switch (signal) {

     

       
225
       
 
       
          | Pull => {

     
···

       
248
       
 
       
    });

     

       
249
       
 
       


     

       
250
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
251
       
+
       
      Wonka_thelpers.testWithListenable(Wonka.filter((._) => true))

     

       
252
       
 
       
        |> Js.Promise.then_(x => {

     

       
253
       
 
       
          expect(x)

     

       
254
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
257
       
 
       
    });

     

       
258
       
 
       


     

       
259
       
 
       
    testPromise("follows the spec for listenables when filtering", () => {

     

       
260
       
+
       
      Wonka_thelpers.testWithListenable(Wonka.filter((._) => false))

     

       
261
       
 
       
        |> Js.Promise.then_(x => {

     

       
262
       
 
       
          expect(x)

     

       
263
       
 
       
            |> toEqual(([| Pull, Pull |], [| End |]))

     
···

       
268
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
269
       
 
       
      let end_: talkbackT = Close;

     

       
270
       
 
       


     

       
271
       
+
       
      Wonka_thelpers.testTalkbackEnd(Wonka.filter((._) => true))

     

       
272
       
 
       
        |> Js.Promise.then_(x => {

     

       
273
       
 
       
          expect(x)

     

       
274
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
742
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
743
       
 
       
      let num = ref(1);

     

       
744
       
 
       


     

       
745
       
+
       
      let source = Wonka.takeWhile((.x) => x <= 2, sink => sink(.Start((.signal) => {

     

       
746
       
 
       
        switch (signal) {

     

       
747
       
 
       
        | Pull => {

     

       
748
       
 
       
          let i = num^;

     
···

       
774
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
775
       
 
       
      let num = ref(1);

     

       
776
       
 
       


     

       
777
       
+
       
      let source = Wonka.takeWhile((.x) => x <= 5, sink => sink(.Start((.signal) => {

     

       
778
       
 
       
        switch (signal) {

     

       
779
       
 
       
        | Pull => {

     

       
780
       
 
       
          let i = num^;

     
···

       
806
       
 
       
    });

     

       
807
       
 
       


     

       
808
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
809
       
+
       
      Wonka_thelpers.testWithListenable(Wonka.takeWhile((._) => true))

     

       
810
       
 
       
        |> Js.Promise.then_(x => {

     

       
811
       
 
       
          expect(x)

     

       
812
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
817
       
 
       
    testPromise("follows the spec for listenables when ending the source", () => {

     

       
818
       
 
       
      let end_: talkbackT = Close;

     

       
819
       
 
       


     

       
820
       
+
       
      Wonka_thelpers.testWithListenable(Wonka.takeWhile((._) => false))

     

       
821
       
 
       
        |> Js.Promise.then_(x => {

     

       
822
       
 
       
          expect(x)

     

       
823
       
 
       
            |> toEqual(([| end_ |], [| End |]))

     
···

       
828
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
829
       
 
       
      let end_: talkbackT = Close;

     

       
830
       
 
       


     

       
831
       
+
       
      Wonka_thelpers.testTalkbackEnd(Wonka.takeWhile((._) => true))

     

       
832
       
 
       
        |> Js.Promise.then_(x => {

     

       
833
       
 
       
          expect(x)

     

       
834
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
1023
       
 
       
      let talkback = ref((._: Wonka_types.talkbackT) => ());

     

       
1024
       
 
       
      let num = ref(1);

     

       
1025
       
 
       


     

       
1026
       
+
       
      let source = Wonka.skipWhile((.x) => x <= 2, sink => sink(.Start((.signal) => {

     

       
1027
       
 
       
        switch (signal) {

     

       
1028
       
 
       
        | Pull when num^ <= 4 => {

     

       
1029
       
 
       
          let i = num^;

     
···

       
1051
       
 
       
    });

     

       
1052
       
 
       


     

       
1053
       
 
       
    testPromise("follows the spec for listenables", () => {

     

       
1054
       
+
       
      Wonka_thelpers.testWithListenable(Wonka.skipWhile((._) => false))

     

       
1055
       
 
       
        |> Js.Promise.then_(x => {

     

       
1056
       
 
       
          expect(x)

     

       
1057
       
 
       
            |> toEqual(([||], [| Push(1), Push(2), End |]))

     
···

       
1060
       
 
       
    });

     

       
1061
       
 
       


     

       
1062
       
 
       
    testPromise("follows the spec for listenables when skipping the source", () => {

     

       
1063
       
+
       
      Wonka_thelpers.testWithListenable(Wonka.skipWhile((._) => true))

     

       
1064
       
 
       
        |> Js.Promise.then_(x => {

     

       
1065
       
 
       
          expect(x)

     

       
1066
       
 
       
            |> toEqual(([| Pull, Pull |], [| End |]))

     
···

       
1071
       
 
       
    testPromise("ends itself and source when its talkback receives the End signal", () => {

     

       
1072
       
 
       
      let end_: talkbackT = Close;

     

       
1073
       
 
       


     

       
1074
       
+
       
      Wonka_thelpers.testTalkbackEnd(Wonka.skipWhile((._) => false))

     

       
1075
       
 
       
        |> Js.Promise.then_(x => {

     

       
1076
       
 
       
          expect(x)

     

       
1077
       
 
       
            |> toEqual(([| end_ |], [| Push(1) |]))

     
···

       
1288
       
 
       


     

       
1289
       
 
       
    input

     

       
1290
       
 
       
      |> Wonka.fromArray

     

       
1291
       
+
       
      |> Wonka.map((.x) => string_of_int(x))

     

       
1292
       
 
       
      |> Wonka.forEach((.x) => ignore(Js.Array.push(x, actual)));

     

       
1293
       
 
       


     

       
1294
       
 
       
    expect(output) |> toEqual(output)

     

···

       
77
       
 
       
  }));

     

       
78
       
 
       
});

     

       
79
       
 
       


     

       
80
       
-
       
let debounce = debounceF => curry(source => curry(sink => {

     

       
81
       
 
       
  let gotEndSignal = ref(false);

     

       
82
       
 
       
  let id: ref(option(Js.Global.timeoutId)) = ref(None);

     

       
83
       
 
       


     
···

       
109
       
 
       
        id := None;

     

       
110
       
 
       
        sink(.signal);

     

       
111
       
 
       
        if (gotEndSignal^) sink(.End);

     

       
112
       
-
       
      }, debounceF(x)));

     

       
113
       
 
       
    }

     

       
114
       
 
       
    | End => {

     

       
115
       
 
       
      gotEndSignal := true;

     
···

       
123
       
 
       
  });

     

       
124
       
 
       
}));

     

       
125
       
 
       


     

       
126
       
-
       
let throttle = throttleF => curry(source => curry(sink => {

     

       
127
       
 
       
  let skip = ref(false);

     

       
128
       
 
       
  let id: ref(option(Js.Global.timeoutId)) = ref(None);

     

       
129
       
 
       
  let clearTimeout = () =>

     
···

       
155
       
 
       
      id := Some(Js.Global.setTimeout(() => {

     

       
156
       
 
       
        id := None;

     

       
157
       
 
       
        skip := false;

     

       
158
       
-
       
      }, throttleF(x)));

     

       
159
       
 
       
      sink(.signal);

     

       
160
       
 
       
    }

     

       
161
       
 
       
    | Push(_) => ()

     

···

       
77
       
 
       
  }));

     

       
78
       
 
       
});

     

       
79
       
 
       


     

       
80
       
+
       
let debounce = f => curry(source => curry(sink => {

     

       
81
       
 
       
  let gotEndSignal = ref(false);

     

       
82
       
 
       
  let id: ref(option(Js.Global.timeoutId)) = ref(None);

     

       
83
       
 
       


     
···

       
109
       
 
       
        id := None;

     

       
110
       
 
       
        sink(.signal);

     

       
111
       
 
       
        if (gotEndSignal^) sink(.End);

     

       
112
       
+
       
      }, f(.x)));

     

       
113
       
 
       
    }

     

       
114
       
 
       
    | End => {

     

       
115
       
 
       
      gotEndSignal := true;

     
···

       
123
       
 
       
  });

     

       
124
       
 
       
}));

     

       
125
       
 
       


     

       
126
       
+
       
let throttle = f => curry(source => curry(sink => {

     

       
127
       
 
       
  let skip = ref(false);

     

       
128
       
 
       
  let id: ref(option(Js.Global.timeoutId)) = ref(None);

     

       
129
       
 
       
  let clearTimeout = () =>

     
···

       
155
       
 
       
      id := Some(Js.Global.setTimeout(() => {

     

       
156
       
 
       
        id := None;

     

       
157
       
 
       
        skip := false;

     

       
158
       
+
       
      }, f(.x)));

     

       
159
       
 
       
      sink(.signal);

     

       
160
       
 
       
    }

     

       
161
       
 
       
    | Push(_) => ()

     

···

       
44
       
 
       
/* Takes a projection to a period in milliseconds and a source, and creates

     

       
45
       
 
       
   a listenable source that emits the last emitted value if no other value

     

       
46
       
 
       
   has been emitted during the passed debounce period. */

     

       
47
       
-
       
let debounce: ('a => int, sourceT('a), sinkT('a)) => unit;

     

       
48
       
 
       


     

       
49
       
 
       
/* Takes a projection to a period in milliseconds and a source, and creates

     

       
50
       
 
       
   a listenable source that ignores values after the last emitted value for

     

       
51
       
 
       
   the duration of the returned throttle period. */

     

       
52
       
-
       
let throttle: ('a => int, sourceT('a), sinkT('a)) => unit;

     

       
53
       
 
       


     

       
54
       
 
       
/* Takes a notifier source and an input source, and creates a sink & source.

     

       
55
       
 
       
   When the notifier emits a value, it will emit the value that it most recently

     
···

       
63
       
 
       


     

       
64
       
 
       
/* Converts a stream into a promise by resolving to the last value of the

     

       
65
       
 
       
   stream. */

     

       
66
       
-
       
let toPromise: (sourceT('a)) => Js.Promise.t('a);

     

···

       
44
       
 
       
/* Takes a projection to a period in milliseconds and a source, and creates

     

       
45
       
 
       
   a listenable source that emits the last emitted value if no other value

     

       
46
       
 
       
   has been emitted during the passed debounce period. */

     

       
47
       
+
       
let debounce: ((.'a) => int, sourceT('a), sinkT('a)) => unit;

     

       
48
       
 
       


     

       
49
       
 
       
/* Takes a projection to a period in milliseconds and a source, and creates

     

       
50
       
 
       
   a listenable source that ignores values after the last emitted value for

     

       
51
       
 
       
   the duration of the returned throttle period. */

     

       
52
       
+
       
let throttle: ((.'a) => int, sourceT('a), sinkT('a)) => unit;

     

       
53
       
 
       


     

       
54
       
 
       
/* Takes a notifier source and an input source, and creates a sink & source.

     

       
55
       
 
       
   When the notifier emits a value, it will emit the value that it most recently

     
···

       
63
       
 
       


     

       
64
       
 
       
/* Converts a stream into a promise by resolving to the last value of the

     

       
65
       
 
       
   stream. */

     

       
66
       
+
       
let toPromise: sourceT('a) => Js.Promise.t('a);

     

···

       
38
       
 
       
};

     

       
39
       
 
       


     

       
40
       
 
       
let make = f => curry(sink => {

     

       
41
       
-
       
  let teardown = f({

     

       
42
       
 
       
    next: value => sink(.Push(value)),

     

       
43
       
 
       
    complete: () => sink(.End)

     

       
44
       
 
       
  });

     

       
45
       
 
       


     

       
46
       
 
       
  sink(.Start((.signal) => {

     

       
47
       
 
       
    switch (signal) {

     

       
48
       
-
       
    | Close => teardown()

     

       
49
       
 
       
    | Pull => ()

     

       
50
       
 
       
    }

     

       
51
       
 
       
  }));

     
···

       
176
       
 
       
  source((.signal) => sink(.

     

       
177
       
 
       
    switch (signal) {

     

       
178
       
 
       
    | Start(x) => Start(x)

     

       
179
       
-
       
    | Push(x) => Push(f(x))

     

       
180
       
 
       
    | End => End

     

       
181
       
 
       
    }

     

       
182
       
 
       
  ));

     
···

       
185
       
 
       
let filter = f => curry(source => curry(sink => {

     

       
186
       
 
       
  captureTalkback(source, (.signal, talkback) => {

     

       
187
       
 
       
    switch (signal) {

     

       
188
       
-
       
    | Push(x) when !f(x) => talkback(.Pull)

     

       
189
       
 
       
    | _ => sink(.signal)

     

       
190
       
 
       
    }

     

       
191
       
 
       
  });

     
···

       
258
       
 
       
      tb(.Pull);

     

       
259
       
 
       
    }

     

       
260
       
 
       
    | Push(x) when !state.ended => {

     

       
261
       
-
       
      applyInnerSource(f(x));

     

       
262
       
 
       
      state.outerTalkback(.Pull);

     

       
263
       
 
       
    }

     

       
264
       
 
       
    | Push(_) => ()

     
···

       
281
       
 
       
  }));

     

       
282
       
 
       
}));

     

       
283
       
 
       


     

       
284
       
-
       
let merge = sources => mergeMap(identity, fromArray(sources));

     

       
285
       
-
       
let mergeAll = source => mergeMap(identity, source);

     

       
286
       
 
       
let flatten = mergeAll;

     

       
287
       
 
       


     

       
288
       
 
       
type concatMapStateT('a) = {

     
···

       
312
       
 
       
        state.innerTalkback = talkbackPlaceholder;

     

       
313
       
 
       


     

       
314
       
 
       
        switch (Rebel.MutableQueue.pop(state.inputQueue)) {

     

       
315
       
-
       
        | Some(input) => applyInnerSource(f(input))

     

       
316
       
 
       
        | None when state.ended => sink(.End)

     

       
317
       
 
       
        | None => ()

     

       
318
       
 
       
        };

     
···

       
347
       
 
       
      if (state.innerActive) {

     

       
348
       
 
       
        Rebel.MutableQueue.add(state.inputQueue, x);

     

       
349
       
 
       
      } else {

     

       
350
       
-
       
        applyInnerSource(f(x));

     

       
351
       
 
       
      }

     

       
352
       
 
       


     

       
353
       
 
       
      state.outerTalkback(.Pull);

     
···

       
370
       
 
       
  }));

     

       
371
       
 
       
}));

     

       
372
       
 
       


     

       
373
       
-
       
let concatAll = source => concatMap(identity, source);

     

       
374
       
-
       
let concat = sources => concatMap(identity, fromArray(sources));

     

       
375
       
 
       


     

       
376
       
 
       
type switchMapStateT('a) = {

     

       
377
       
 
       
  mutable outerTalkback: (.talkbackT) => unit,

     
···

       
427
       
 
       
        state.innerTalkback(.Close);

     

       
428
       
 
       
        state.innerTalkback = talkbackPlaceholder;

     

       
429
       
 
       
      }

     

       
430
       
-
       
      applyInnerSource(f(x));

     

       
431
       
 
       
      state.outerTalkback(.Pull);

     

       
432
       
 
       
    }

     

       
433
       
 
       
    | Push(_) => ()

     
···

       
649
       
 
       
  });

     

       
650
       
 
       
}));

     

       
651
       
 
       


     

       
652
       
-
       
let takeWhile = predicate => curry(source => curry(sink => {

     

       
653
       
 
       
  let ended = ref(false);

     

       
654
       
 
       
  let talkback = ref(talkbackPlaceholder);

     

       
655
       
 
       


     
···

       
665
       
 
       
    }

     

       
666
       
 
       
    | End => ()

     

       
667
       
 
       
    | Push(x) when !ended^ => {

     

       
668
       
-
       
      if (!predicate(x)) {

     

       
669
       
 
       
        ended := true;

     

       
670
       
 
       
        sink(.End);

     

       
671
       
 
       
        talkback^(.Close);

     
···

       
762
       
 
       
  });

     

       
763
       
 
       
}));

     

       
764
       
 
       


     

       
765
       
-
       
let skipWhile = predicate => curry(source => curry(sink => {

     

       
766
       
 
       
  let skip = ref(true);

     

       
767
       
 
       


     

       
768
       
 
       
  captureTalkback(source, (.signal, talkback) => {

     

       
769
       
 
       
    switch (signal) {

     

       
770
       
 
       
    | Push(x) when skip^ => {

     

       
771
       
-
       
      if (predicate(x)) {

     

       
772
       
 
       
        talkback(.Pull);

     

       
773
       
 
       
      } else {

     

       
774
       
 
       
        skip := false;

     

···

       
38
       
 
       
};

     

       
39
       
 
       


     

       
40
       
 
       
let make = f => curry(sink => {

     

       
41
       
+
       
  let teardown = f(.{

     

       
42
       
 
       
    next: value => sink(.Push(value)),

     

       
43
       
 
       
    complete: () => sink(.End)

     

       
44
       
 
       
  });

     

       
45
       
 
       


     

       
46
       
 
       
  sink(.Start((.signal) => {

     

       
47
       
 
       
    switch (signal) {

     

       
48
       
+
       
    | Close => teardown(.)

     

       
49
       
 
       
    | Pull => ()

     

       
50
       
 
       
    }

     

       
51
       
 
       
  }));

     
···

       
176
       
 
       
  source((.signal) => sink(.

     

       
177
       
 
       
    switch (signal) {

     

       
178
       
 
       
    | Start(x) => Start(x)

     

       
179
       
+
       
    | Push(x) => Push(f(.x))

     

       
180
       
 
       
    | End => End

     

       
181
       
 
       
    }

     

       
182
       
 
       
  ));

     
···

       
185
       
 
       
let filter = f => curry(source => curry(sink => {

     

       
186
       
 
       
  captureTalkback(source, (.signal, talkback) => {

     

       
187
       
 
       
    switch (signal) {

     

       
188
       
+
       
    | Push(x) when !f(.x) => talkback(.Pull)

     

       
189
       
 
       
    | _ => sink(.signal)

     

       
190
       
 
       
    }

     

       
191
       
 
       
  });

     
···

       
258
       
 
       
      tb(.Pull);

     

       
259
       
 
       
    }

     

       
260
       
 
       
    | Push(x) when !state.ended => {

     

       
261
       
+
       
      applyInnerSource(f(.x));

     

       
262
       
 
       
      state.outerTalkback(.Pull);

     

       
263
       
 
       
    }

     

       
264
       
 
       
    | Push(_) => ()

     
···

       
281
       
 
       
  }));

     

       
282
       
 
       
}));

     

       
283
       
 
       


     

       
284
       
+
       
let merge = sources => mergeMap((.x) => x, fromArray(sources));

     

       
285
       
+
       
let mergeAll = source => mergeMap((.x) => x, source);

     

       
286
       
 
       
let flatten = mergeAll;

     

       
287
       
 
       


     

       
288
       
 
       
type concatMapStateT('a) = {

     
···

       
312
       
 
       
        state.innerTalkback = talkbackPlaceholder;

     

       
313
       
 
       


     

       
314
       
 
       
        switch (Rebel.MutableQueue.pop(state.inputQueue)) {

     

       
315
       
+
       
        | Some(input) => applyInnerSource(f(.input))

     

       
316
       
 
       
        | None when state.ended => sink(.End)

     

       
317
       
 
       
        | None => ()

     

       
318
       
 
       
        };

     
···

       
347
       
 
       
      if (state.innerActive) {

     

       
348
       
 
       
        Rebel.MutableQueue.add(state.inputQueue, x);

     

       
349
       
 
       
      } else {

     

       
350
       
+
       
        applyInnerSource(f(.x));

     

       
351
       
 
       
      }

     

       
352
       
 
       


     

       
353
       
 
       
      state.outerTalkback(.Pull);

     
···

       
370
       
 
       
  }));

     

       
371
       
 
       
}));

     

       
372
       
 
       


     

       
373
       
+
       
let concatAll = source => concatMap((.x) => x, source);

     

       
374
       
+
       
let concat = sources => concatMap((.x) => x, fromArray(sources));

     

       
375
       
 
       


     

       
376
       
 
       
type switchMapStateT('a) = {

     

       
377
       
 
       
  mutable outerTalkback: (.talkbackT) => unit,

     
···

       
427
       
 
       
        state.innerTalkback(.Close);

     

       
428
       
 
       
        state.innerTalkback = talkbackPlaceholder;

     

       
429
       
 
       
      }

     

       
430
       
+
       
      applyInnerSource(f(.x));

     

       
431
       
 
       
      state.outerTalkback(.Pull);

     

       
432
       
 
       
    }

     

       
433
       
 
       
    | Push(_) => ()

     
···

       
649
       
 
       
  });

     

       
650
       
 
       
}));

     

       
651
       
 
       


     

       
652
       
+
       
let takeWhile = f => curry(source => curry(sink => {

     

       
653
       
 
       
  let ended = ref(false);

     

       
654
       
 
       
  let talkback = ref(talkbackPlaceholder);

     

       
655
       
 
       


     
···

       
665
       
 
       
    }

     

       
666
       
 
       
    | End => ()

     

       
667
       
 
       
    | Push(x) when !ended^ => {

     

       
668
       
+
       
      if (!f(.x)) {

     

       
669
       
 
       
        ended := true;

     

       
670
       
 
       
        sink(.End);

     

       
671
       
 
       
        talkback^(.Close);

     
···

       
762
       
 
       
  });

     

       
763
       
 
       
}));

     

       
764
       
 
       


     

       
765
       
+
       
let skipWhile = f => curry(source => curry(sink => {

     

       
766
       
 
       
  let skip = ref(true);

     

       
767
       
 
       


     

       
768
       
 
       
  captureTalkback(source, (.signal, talkback) => {

     

       
769
       
 
       
    switch (signal) {

     

       
770
       
 
       
    | Push(x) when skip^ => {

     

       
771
       
+
       
      if (f(.x)) {

     

       
772
       
 
       
        talkback(.Pull);

     

       
773
       
 
       
      } else {

     

       
774
       
 
       
        skip := false;

     

···

       
13
       
 
       
   for constructing any kind of asynchronous stream. The return

     

       
14
       
 
       
   callback from the passed observer function will be called when

     

       
15
       
 
       
   the stream is closed or ends */

     

       
16
       
-
       
let make: (observerT('a) => (unit => unit), sinkT('a)) => unit;

     

       
17
       
 
       


     

       
18
       
 
       
/* Accepts a list and creates a pullable source for that list.

     

       
19
       
 
       
   The source will emit events when being pulled until the list

     
···

       
45
       
 
       
   and creates a sink & source.

     

       
46
       
 
       
   All values that it receives will be transformed using the mapping

     

       
47
       
 
       
   function and emitted on the new source */

     

       
48
       
-
       
let map: ('a => 'b, sourceT('a), sinkT('b)) => unit;

     

       
49
       
 
       


     

       
50
       
 
       
/* Takes a predicate function returning a boolean, and a source,

     

       
51
       
 
       
   and creates a sink & source.

     

       
52
       
 
       
   All values that it receives will be filtered using the predicate,

     

       
53
       
 
       
   and only truthy values will be passed on to the new source.

     

       
54
       
 
       
   The sink will attempt to pull a new value when one was filtered. */

     

       
55
       
-
       
let filter: ('a => bool, sourceT('a), sinkT('a)) => unit;

     

       
56
       
 
       


     

       
57
       
 
       
/* Takes a reducer function, a seed value, and a source, and creates

     

       
58
       
 
       
   a sink & source.

     
···

       
66
       
 
       
   and a source, and creates a sink & source.

     

       
67
       
 
       
   The mapping function is called with each value it receives and

     

       
68
       
 
       
   the resulting inner source is merged into the output source. */

     

       
69
       
-
       
let mergeMap: ('a => sourceT('b), sourceT('a), sinkT('b)) => unit;

     

       
70
       
 
       


     

       
71
       
 
       
/* Takes a mapping function from one types to a source output,

     

       
72
       
 
       
   and a source, and creates a sink & source.

     

       
73
       
 
       
   The mapping function is called with each value it receives and

     

       
74
       
 
       
   the latest inner source is merged into the output source. When

     

       
75
       
 
       
   a new value comes in the previous source is dicarded. */

     

       
76
       
-
       
let switchMap: ('a => sourceT('b), sourceT('a), sinkT('b)) => unit;

     

       
77
       
 
       


     

       
78
       
 
       
/* Takes a mapping function from one types to a source output,

     

       
79
       
 
       
   and a source, and creates a sink & source.

     

       
80
       
 
       
   The mapping function is called with each value it receives and

     

       
81
       
 
       
   the resulting inner sources are subscribed to and piped through

     

       
82
       
 
       
   to the output source in order. */

     

       
83
       
-
       
let concatMap: ('a => sourceT('b), sourceT('a), sinkT('b)) => unit;

     

       
84
       
 
       


     

       
85
       
 
       
/* Takes an array of sources and creates a sink & source.

     

       
86
       
 
       
   All values that the sink receives from all sources will be passed through

     
···

       
127
       
 
       
   It will emit values that the sink receives until the predicate returns false

     

       
128
       
 
       
   for a value, at which point it will end the source and the returned, new

     

       
129
       
 
       
   source. */

     

       
130
       
-
       
let takeWhile: ('a => bool, sourceT('a), sinkT('a)) => unit;

     

       
131
       
 
       


     

       
132
       
 
       
/* Takes a notifier source and an input source, and creates a sink & source.

     

       
133
       
 
       
   It will not emit values that the sink receives until the notifier source

     
···

       
150
       
 
       
   It will not emit values that the sink receives until the passed predicate

     

       
151
       
 
       
   returns false for a value, at which point it will start acting like a noop

     

       
152
       
 
       
   operator, passing through every signal. */

     

       
153
       
-
       
let skipWhile: ('a => bool, sourceT('a), sinkT('a)) => unit;

     

       
154
       
 
       


     

       
155
       
 
       
/* Takes a notifier source and an input source, and creates a sink & source.

     

       
156
       
 
       
   It will not emit values that the sink receives until the notifier source

     

···

       
13
       
 
       
   for constructing any kind of asynchronous stream. The return

     

       
14
       
 
       
   callback from the passed observer function will be called when

     

       
15
       
 
       
   the stream is closed or ends */

     

       
16
       
+
       
let make: ((.observerT('a)) => teardownT, sinkT('a)) => unit;

     

       
17
       
 
       


     

       
18
       
 
       
/* Accepts a list and creates a pullable source for that list.

     

       
19
       
 
       
   The source will emit events when being pulled until the list

     
···

       
45
       
 
       
   and creates a sink & source.

     

       
46
       
 
       
   All values that it receives will be transformed using the mapping

     

       
47
       
 
       
   function and emitted on the new source */

     

       
48
       
+
       
let map: ((.'a) => 'b, sourceT('a), sinkT('b)) => unit;

     

       
49
       
 
       


     

       
50
       
 
       
/* Takes a predicate function returning a boolean, and a source,

     

       
51
       
 
       
   and creates a sink & source.

     

       
52
       
 
       
   All values that it receives will be filtered using the predicate,

     

       
53
       
 
       
   and only truthy values will be passed on to the new source.

     

       
54
       
 
       
   The sink will attempt to pull a new value when one was filtered. */

     

       
55
       
+
       
let filter: ((.'a) => bool, sourceT('a), sinkT('a)) => unit;

     

       
56
       
 
       


     

       
57
       
 
       
/* Takes a reducer function, a seed value, and a source, and creates

     

       
58
       
 
       
   a sink & source.

     
···

       
66
       
 
       
   and a source, and creates a sink & source.

     

       
67
       
 
       
   The mapping function is called with each value it receives and

     

       
68
       
 
       
   the resulting inner source is merged into the output source. */

     

       
69
       
+
       
let mergeMap: ((.'a) => sourceT('b), sourceT('a), sinkT('b)) => unit;

     

       
70
       
 
       


     

       
71
       
 
       
/* Takes a mapping function from one types to a source output,

     

       
72
       
 
       
   and a source, and creates a sink & source.

     

       
73
       
 
       
   The mapping function is called with each value it receives and

     

       
74
       
 
       
   the latest inner source is merged into the output source. When

     

       
75
       
 
       
   a new value comes in the previous source is dicarded. */

     

       
76
       
+
       
let switchMap: ((.'a) => sourceT('b), sourceT('a), sinkT('b)) => unit;

     

       
77
       
 
       


     

       
78
       
 
       
/* Takes a mapping function from one types to a source output,

     

       
79
       
 
       
   and a source, and creates a sink & source.

     

       
80
       
 
       
   The mapping function is called with each value it receives and

     

       
81
       
 
       
   the resulting inner sources are subscribed to and piped through

     

       
82
       
 
       
   to the output source in order. */

     

       
83
       
+
       
let concatMap: ((.'a) => sourceT('b), sourceT('a), sinkT('b)) => unit;

     

       
84
       
 
       


     

       
85
       
 
       
/* Takes an array of sources and creates a sink & source.

     

       
86
       
 
       
   All values that the sink receives from all sources will be passed through

     
···

       
127
       
 
       
   It will emit values that the sink receives until the predicate returns false

     

       
128
       
 
       
   for a value, at which point it will end the source and the returned, new

     

       
129
       
 
       
   source. */

     

       
130
       
+
       
let takeWhile: ((.'a) => bool, sourceT('a), sinkT('a)) => unit;

     

       
131
       
 
       


     

       
132
       
 
       
/* Takes a notifier source and an input source, and creates a sink & source.

     

       
133
       
 
       
   It will not emit values that the sink receives until the notifier source

     
···

       
150
       
 
       
   It will not emit values that the sink receives until the passed predicate

     

       
151
       
 
       
   returns false for a value, at which point it will start acting like a noop

     

       
152
       
 
       
   operator, passing through every signal. */

     

       
153
       
+
       
let skipWhile: ((.'a) => bool, sourceT('a), sinkT('a)) => unit;

     

       
154
       
 
       


     

       
155
       
 
       
/* Takes a notifier source and an input source, and creates a sink & source.

     

       
156
       
 
       
   It will not emit values that the sink receives until the notifier source

     

···

       
1
       
 
       
open Wonka_types;

     

       
2
       
 
       


     

       
3
       
-
       
external identity: 'a => 'a = "%identity";

     

       
4
       
-
       


     

       
5
       
 
       
let talkbackPlaceholder = (._: talkbackT) => ();

     

       
6
       
 
       


     

       
7
       
 
       
let captureTalkback = (

     

···

       
1
       
 
       
open Wonka_types;

     

       
2
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
3
       
 
       
let talkbackPlaceholder = (._: talkbackT) => ();

     

       
4
       
 
       


     

       
5
       
 
       
let captureTalkback = (

     

···

       
38
       
 
       
type sinkT('a) = (.signalT('a)) => unit;

     

       
39
       
 
       
type sourceT('a) = sinkT('a) => unit;

     

       
40
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
41
       
 
       
type subscriptionT = {

     

       
42
       
 
       
  unsubscribe: unit => unit

     

       
43
       
 
       
};

     

···

       
38
       
 
       
type sinkT('a) = (.signalT('a)) => unit;

     

       
39
       
 
       
type sourceT('a) = sinkT('a) => unit;

     

       
40
       
 
       


     

       
41
       
+
       
type teardownT = (.unit) => unit;

     

       
42
       
+
       


     

       
43
       
 
       
type subscriptionT = {

     

       
44
       
 
       
  unsubscribe: unit => unit

     

       
45
       
 
       
};