···

       
1242
       
1242
       
 
       
        }));

     

       
1243
       
1243
       
 
       
      };

     

       
1244
       
1244
       
 
       


     

       
1245
       
1245
       
-
       
      Wonka.forEach(x => ignore(Js.Array.push(x, nums)), source);

     

       
1245
       
1245
       
+
       
      Wonka.forEach((.x) => ignore(Js.Array.push(x, nums)), source);

     

       
1246
       
1246
       
 
       
      expect(nums) |> toEqual([| 0, 1, 2, 3 |])

     

       
1247
       
1247
       
 
       
    });

     

       
1248
       
1248
       
 
       
  });

     
···

       
1265
       
1265
       
 
       
        sink(.Start(Wonka_helpers.talkbackPlaceholder));

     

       
1266
       
1266
       
 
       
      };

     

       
1267
       
1267
       
 
       


     

       
1268
       
1268
       
-
       
      let { unsubscribe } = Wonka.subscribe(x => ignore(Js.Array.push(x, nums)), source);

     

       
1268
       
1268
       
+
       
      let { unsubscribe } = Wonka.subscribe((.x) => ignore(Js.Array.push(x, nums)), source);

     

       
1269
       
1269
       
 
       


     

       
1270
       
1270
       
 
       
      push^();

     

       
1271
       
1271
       
 
       
      push^();

     
···

       
1289
       
1289
       
 
       
    input

     

       
1290
       
1290
       
 
       
      |> Wonka.fromArray

     

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

     

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

     

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

     

       
1293
       
1293
       
 
       


     

       
1294
       
1294
       
 
       
    expect(output) |> toEqual(output)

     

       
1295
       
1295
       
 
       
  });

     

···

       
35
       
35
       
 
       
export const skipWhile: <A>(f: (x: A) => boolean) => Operator<A, A>;

     

       
36
       
36
       
 
       
export const skipUntil: <A>(signal: Source<any>) => Operator<A, A>;

     

       
37
       
37
       
 
       


     

       
38
       
38
       
+
       
export const publish: <A>(source: Source<A>) => Subscription;

     

       
38
       
39
       
 
       
export const forEach: <A>(f: (x: A) => void) => (source: Source<A>) => void;

     

       
39
       
40
       
 
       
export const subscribe: <A>(f: (x: A) => void) => (source: Source<A>) => Subscription;

     

···

       
164
       
164
       
 
       
let tap = f => curry(source => curry(sink => {

     

       
165
       
165
       
 
       
  source((.signal) => {

     

       
166
       
166
       
 
       
    switch (signal) {

     

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

     

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

     

       
168
       
168
       
 
       
    | _ => ()

     

       
169
       
169
       
 
       
    };

     

       
170
       
170
       
 
       


     
···

       
847
       
847
       
 
       
  }));

     

       
848
       
848
       
 
       
}));

     

       
849
       
849
       
 
       


     

       
850
       
850
       
-
       
let forEach = f => curry(source => {

     

       
851
       
851
       
-
       
  captureTalkback(source, (.signal, talkback) => {

     

       
850
       
850
       
+
       
type publishStateT = {

     

       
851
       
851
       
+
       
  mutable talkback: (.talkbackT) => unit,

     

       
852
       
852
       
+
       
  mutable ended: bool

     

       
853
       
853
       
+
       
};

     

       
854
       
854
       
+
       


     

       
855
       
855
       
+
       
let publish = source => {

     

       
856
       
856
       
+
       
  let state: publishStateT = {

     

       
857
       
857
       
+
       
    talkback: talkbackPlaceholder,

     

       
858
       
858
       
+
       
    ended: false

     

       
859
       
859
       
+
       
  };

     

       
860
       
860
       
+
       


     

       
861
       
861
       
+
       
  source((.signal) => {

     

       
852
       
862
       
 
       
    switch (signal) {

     

       
853
       
853
       
-
       
    | Start(_) => talkback(.Pull)

     

       
854
       
854
       
-
       
    | Push(x) => {

     

       
855
       
855
       
-
       
      f(x);

     

       
856
       
856
       
-
       
      talkback(.Pull);

     

       
863
       
863
       
+
       
    | Start(x) => {

     

       
864
       
864
       
+
       
      state.talkback = x;

     

       
865
       
865
       
+
       
      x(.Pull);

     

       
857
       
866
       
 
       
    }

     

       
858
       
858
       
-
       
    | End => ()

     

       
867
       
867
       
+
       
    | Push(_) => if (!state.ended) state.talkback(.Pull);

     

       
868
       
868
       
+
       
    | End => state.ended = true;

     

       
859
       
869
       
 
       
    }

     

       
860
       
870
       
 
       
  });

     

       
861
       
861
       
-
       
});

     

       
871
       
871
       
+
       


     

       
872
       
872
       
+
       
  {

     

       
873
       
873
       
+
       
    unsubscribe: () =>

     

       
874
       
874
       
+
       
      if (!state.ended) {

     

       
875
       
875
       
+
       
        state.ended = true;

     

       
876
       
876
       
+
       
        state.talkback(.Close);

     

       
877
       
877
       
+
       
      }

     

       
878
       
878
       
+
       
  }

     

       
879
       
879
       
+
       
};

     

       
862
       
880
       
 
       


     

       
863
       
881
       
 
       
let subscribe = f => curry(source => {

     

       
864
       
864
       
-
       
  let talkback = ref(talkbackPlaceholder);

     

       
865
       
865
       
-
       
  let ended = ref(false);

     

       
882
       
882
       
+
       
  let state: publishStateT = {

     

       
883
       
883
       
+
       
    talkback: talkbackPlaceholder,

     

       
884
       
884
       
+
       
    ended: false

     

       
885
       
885
       
+
       
  };

     

       
866
       
886
       
 
       


     

       
867
       
887
       
 
       
  source((.signal) => {

     

       
868
       
888
       
 
       
    switch (signal) {

     

       
869
       
889
       
 
       
    | Start(x) => {

     

       
870
       
870
       
-
       
      talkback := x;

     

       
890
       
890
       
+
       
      state.talkback = x;

     

       
871
       
891
       
 
       
      x(.Pull);

     

       
872
       
892
       
 
       
    }

     

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

     

       
874
       
874
       
-
       
      f(x);

     

       
875
       
875
       
-
       
      talkback^(.Pull);

     

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

     

       
894
       
894
       
+
       
      f(.x);

     

       
895
       
895
       
+
       
      state.talkback(.Pull);

     

       
876
       
896
       
 
       
    }

     

       
877
       
877
       
-
       
    | _ => ()

     

       
897
       
897
       
+
       
    | Push(_) => ()

     

       
898
       
898
       
+
       
    | End => state.ended = true;

     

       
878
       
899
       
 
       
    }

     

       
879
       
900
       
 
       
  });

     

       
880
       
901
       
 
       


     

       
881
       
902
       
 
       
  {

     

       
882
       
903
       
 
       
    unsubscribe: () =>

     

       
883
       
883
       
-
       
      if (!ended^) {

     

       
884
       
884
       
-
       
        ended := true;

     

       
885
       
885
       
-
       
        talkback^(.Close);

     

       
904
       
904
       
+
       
      if (!state.ended) {

     

       
905
       
905
       
+
       
        state.ended = true;

     

       
906
       
906
       
+
       
        state.talkback(.Close);

     

       
886
       
907
       
 
       
      }

     

       
887
       
908
       
 
       
  }

     

       
888
       
909
       
 
       
});

     

       
910
       
910
       
+
       


     

       
911
       
911
       
+
       
let forEach = f => curry(source => {

     

       
912
       
912
       
+
       
  ignore(subscribe(f, source));

     

       
913
       
913
       
+
       
});

     

···

       
39
       
39
       
 
       


     

       
40
       
40
       
 
       
/* Takes a callback and a source, and creates a sink & source.

     

       
41
       
41
       
 
       
   The callback will be called for each value that it receives */

     

       
42
       
42
       
-
       
let tap: ('a => unit, sourceT('a), sinkT('a)) => unit;

     

       
42
       
42
       
+
       
let tap: ((.'a) => unit, sourceT('a), sinkT('a)) => unit;

     

       
43
       
43
       
 
       


     

       
44
       
44
       
 
       
/* Takes a mapping function from one type to another, and a source,

     

       
45
       
45
       
 
       
   and creates a sink & source.

     
···

       
160
       
160
       
 
       


     

       
161
       
161
       
 
       
/* -- sink factories */

     

       
162
       
162
       
 
       


     

       
163
       
163
       
+
       
/* Accepts a source and returns a subscription, but does otherwise not surface values */

     

       
164
       
164
       
+
       
let publish: sourceT('a) => subscriptionT;

     

       
165
       
165
       
+
       


     

       
163
       
166
       
 
       
/* Takes a function and a source, and creates a sink.

     

       
164
       
167
       
 
       
   The function will be called for each value that the sink receives.

     

       
165
       
168
       
 
       
   The sink will attempt to pull new values as values come in, until

     

       
166
       
169
       
 
       
   the source ends. */

     

       
167
       
167
       
-
       
let forEach: ('a => unit, sourceT('a)) => unit;

     

       
170
       
170
       
+
       
let forEach: ((.'a) => unit, sourceT('a)) => unit;

     

       
168
       
171
       
 
       


     

       
169
       
172
       
 
       
/* Similar to the `forEach` sink factory, but returns an anonymous function

     

       
170
       
173
       
 
       
   that when called will end the stream immediately.

     

       
171
       
174
       
 
       
   Ending the stream will propagate an End signal upwards to the root source. */

     

       
172
       
172
       
-
       
let subscribe: ('a => unit, sourceT('a)) => subscriptionT;

     

       
175
       
175
       
+
       
let subscribe: ((.'a) => unit, sourceT('a)) => subscriptionT;