0no-co / wonka

🎩 A tiny but capable push & pull stream library for TypeScript and Flow

Wonka

A tiny but capable push & pull stream library for TypeScript and Flow, loosely following the callbag spec

NOTE: The currently released version v6 is only compatible now with TypeScript, Flow, and JavaScript. If you're looking for Reason/OCaml/esy/dune support, please check v5, and if you're looking for the legacy version of this library check v4.

“There’s no earthly way of knowing
Which direction we are going
There’s no knowing where we’re rowing
Or which way the river’s flowing” － Willy Wonka

Wonka is a lightweight iterable and observable library loosely based on the callbag spec. It exposes a set of helpers to create streams, which are sources of multiple values, which allow you to create, transform and consume event streams or iterable sets of data.

Documentation

See the documentation at wonka.kitten.sh for more information about using wonka!

• Introduction
• Getting started
• Basics
• API Reference

The raw markdown files can be found in this repository in the docs folder.

wonka's Issues

`concatMap` omits values that `mergeMap` catches

Hi There,

I've added two extensions to Wonka that potentially highlight an incompatibility between concatMap and mergeMap.

iterate
A new source that represents a potentially infinite generator from a starting value and an iteration step. Analogous to Haskell's iterate (https://hackage.haskell.org/package/base-4.14.0.0/docs/Prelude.html#v:iterate) but with optional termination.

let iterate =
    (start: 'a, next: 'a => option('a)): Wonka.Types.sourceT(int) => {
  Wonka.Types.curry(sink => {
    let state: Wonka.trampolineT(option('a)) = {
      ended: false,
      looping: false,
      pulled: false,
      current: Some(start),
    };

    sink(.
      Wonka.Types.Start(
        (. signal) =>
          switch (signal, state.looping) {
          | (Pull, false) =>
            state.pulled = true;
            state.looping = true;

            while (state.pulled && !state.ended) {
              switch (state.current) {
              | Some(x) =>
                state.current = next(x);
                state.pulled = false;
                sink(. Push(x));
              | None =>
                state.ended = true;
                sink(. End);
              };
            };
          | (Pull, true) => state.pulled = true
          | (Close, _) => state.ended = true
          },
      ),
    );
  });
};

downTo
An operator for getting numbers start, start - 1, ... down to a provided bound. E.g., downTo(3, 1) would produce [|3, 2, 1|] in a stream.

let downTo = (start: int, bound: int): Wonka.Types.sourceT(int) =>
  iterate(start, x => x <= bound ? None : Some(x - 1));

toArrayAsync
A sink that captures all stream values in an array, wrapped up in a promise. This fixes toArray to work with asynchronous streams.

let toAsyncArray = (source: Wonka.Types.sourceT('a)): Js.Promise.t(array('a)) =>
  Js.Promise.make((~resolve, ~reject as _) => {
    let state: Wonka.toArrayStateT('a) = {
      values: Rebel.MutableQueue.make(),
      talkback: (. _) => (),
      value: None,
      ended: false,
    };

    source((. signal) =>
      switch (signal) {
      | Start(x) =>
        state.talkback = x;
        x(. Pull);
      | Push(value) =>
        Rebel.MutableQueue.add(state.values, value);
        state.talkback(. Pull);
      | End =>
        state.ended = true;
        state.talkback(. Close);
        let finalArray = Rebel.MutableQueue.toArray(state.values);
        resolve(. finalArray);
      }
    );

    ();
  });

Some tests to demonstrate how these work:

test("Wonka iterate down.", () =>
  Library.Wonkatools.downTo(3, 0)
  |> Wonka.toArray
  |> expect == [|3, 2, 1, 0|]
);

test("Wonka sync arrays dont collect", () =>
  [|1, 2, 3|]
  |> Wonka.fromArray
  |> Wonka.mergeMap((. x) => Js.Promise.resolve(x) |> Wonka.fromPromise)
  |> Wonka.toArray
  |> expect == [||]
);

testPromise("Wonka async arrays do collect", () =>
  [|1, 2, 3|]
  |> Wonka.fromArray
  |> Wonka.mergeMap((. x) => Js.Promise.resolve(x) |> Wonka.fromPromise)
  |> Library.Wonkatools.toAsyncArray
  |> Js.Promise.then_(array => array |> expect == [|1, 2, 3|] |> Js.Promise.resolve)
);

The issue

Finally, here is a description of the issue. Below is a test that times out (concatMap fails to collect any values). It times out:

Timeout - Async callback was not invoked within the 5000ms timeout specified by jest.setTimeout.Timeout - Async callback was not invoked within the 5000ms

If I replace concatMap with mergeMap, this test succeeds.

testPromise("Wonka concat problem", () =>
  Library.Wonkatools.downTo(3, 0)
  |> Wonka.concatMap((. x) => Js.Promise.resolve(x) |> Wonka.fromPromise)
  |> Library.Wonkatools.toAsyncArray
  |> Js.Promise.then_(array => array |> expect == [|3, 2, 1, 0|] |> Js.Promise.resolve)
);

Any thoughts on why concatMap is skipping these values?

Tests use bs-jest.

Converting to/from JS Observables

Hi there! As urql is using wonka now, it would be of great help for those of us who want to tinker with it if there was an easy way to pipe to/from RxJS observables. What would be the best approach here?

Wonka module not found

I have been trying to install Wonka on a reason project for last two days and cant seem to get it to be recognized.

reproduction here: https://github.com/idkjs/wonka-not-found

Steps to repoduce error

Setup Project

Dependencies have been installed with pnpm.

pnmp i bs-webapi wonka

Add new deps to bsconfig.json:

  "bs-dependencies": [
    "wonka",
    "reason-react",
    "bs-webapi"
  ],

Run Project

npm install
npm start

On running npm start compiler error is:

➜  wonka-not-found npm start

> [email protected] start /Users/prisc_000/Downloads/wonka-not-found
> bsb -make-world -w -ws _

bsb: no work to do.
bsb: no work to do.
File "bsconfig.json", line 1
Error: package gentype not found or built
- Did you install it?
- If you did, did you run `bsb -make-world`?
>>>> Start compiling
[18/25] Building src/Blinking...ClickEventDemo-ReasonReactExamples.cmj
FAILED: src/BlinkingGreeting/WonkaClickEventDemo-ReasonReactExamples.cmj src/BlinkingGreeting/WonkaClickEventDemo-ReasonReactExamples.cmi /Users/prisc_000/Downloads/wonka-not-found/src/BlinkingGreeting/WonkaClickEventDemo.bs.js
/Users/prisc_000/.fnm/node-versions/v13.3.0/installation/pnpm-global/3/node_modules/.pnpm/registry.npmjs.org/bs-platform/7.0.2-dev.1/node_modules/bs-platform/lib/bsc.exe -nostdlib -bs-package-name reason-react-examples -bs-ns ReasonReactExamples  -bs-package-output commonjs:src/BlinkingGreeting -color always -bs-suffix -I . -I src/ReducerFromReactJSDocs -I src/ReasonUsingJSUsingReason -I src/BlinkingGreeting -I src/FetchedDogPictures -I src -I /Users/prisc_000/Downloads/wonka-not-found/node_modules/reason-react/lib/ocaml -I /Users/prisc_000/Downloads/wonka-not-found/node_modules/bs-webapi/lib/ocaml -I /Users/prisc_000/Downloads/wonka-not-found/node_modules/wonka/lib/ocaml -I /Users/prisc_000/Downloads/wonka-not-found/node_modules/bs-platform/lib/ocaml -w -30-40+6+7+27+32..39+44+45+101 -bs-super-errors -bs-no-version-header -o src/BlinkingGreeting/WonkaClickEventDemo-ReasonReactExamples.cmj src/BlinkingGreeting/WonkaClickEventDemo.reast

  We've found a bug for you!
  /Users/prisc_000/Downloads/wonka-not-found/src/BlinkingGreeting/WonkaClickEventDemo.re 5:6-10

  3 │    fromEvent(document, 'click').subscribe(() => console.log('Clicked!')
      ); */
  4 │ open Webapi.Dom;
  5 │ open Wonka;
  6 │ [@react.component]
  7 │ let make = () => {

  The module or file Wonka can't be found.
  - If it's a third-party dependency:
    - Did you list it in bsconfig.json?
    - Did you run `bsb` instead of `bsb -make-world`
      (latter builds third-parties)?
  - Did you include the file's directory in bsconfig.json?

[23/25] Building src/FetchedD...hedDogPictures-ReasonReactExamples.cmj
FAILED: subcommand failed.
>>>> Finish compiling(exit: 1)

Thank you, sir.

Cannot reference operators, sources, sinks without opening Wonka.

In the upgrade from v2 to v3, we lost the ability to just reference wonka sources, sinks, and operators directly (i.e. as Wonka.subscribe or Wonka.map). Instead, it now only works if you open Wonka at the top of your .re file and additionally reference sources, sinks, and operators in their Wonka.<name> form.

Reproduction

1. Init a new project: bsb -init wonka-test -theme basic-reason (this assumes you already have bs-platform installed globally).
2. Paste this code in the root Reason file (I believe it's Demo.re in the boilerplate):

let source = Wonka.fromArray([|1, 2, 3|]);

source
|> Wonka.map((. _val) => Wonka.interval(_val * 1000) |> Wonka.take(3))
|> Wonka.concatAll
|> Wonka.subscribe((. _val) => print_int(_val));

3. Run yarn build.
4. You should see error output like the following:

➜  wonka-test yarn build
yarn run v1.16.0
$ bsb -make-world
[1/1] Building Wonka.cmi
ninja: no work to do.
[1/1] Building src/Demo-WonkaTest.cmj

  We've found a bug for you!
  /wonka-test/src/Demo.re 1:14-28
  
  1 │ let source = Wonka.fromArray([|1, 2, 3|]);
  2 │ 
  3 │ source
  
  The value fromArray can't be found in Wonka

Similarly, if you just open Wonka and don't reference the module further, you'll get an error. For example, following steps 3 and 4 above with this code:

open Wonka;
let source = fromArray([|1, 2, 3|]);

source
|> map((. _val) => interval(_val * 1000) |> take(3))
|> concatAll
|> subscribe((. _val) => print_int(_val));

will yield the following error:

➜  wonka-test yarn build
yarn run v1.16.0
$ bsb -make-world
[1/1] Building Wonka.cmi
ninja: no work to do.
[1/1] Building src/Demo-WonkaTest.cmj

  We've found a bug for you!
  /Users/parkerziegler/Documents/repos/OSS/wonka-test/src/Demo.re 2:14-22
  
  1 │ open Wonka;
  2 │ let source = fromArray([|1, 2, 3|]);
  3 │ 
  4 │ source
  
  The value fromArray can't be found

I'm thinking there's something about how modules are included that's going awry here, but I'm not 100% sure what it is. But I'll do some investigating when I can and try to see if I can figure out a solution. IIRC this worked fine in v2 so if we need to move back to a structure like that we maybe could? Not sure yet how that affects native stuff.

a is not a function

I'm running wonka within my application, and occasionally this error pops up and crashes my app running locally:

////node_modules/wonka/dist/wonka.js:608
          c.forEach(function (c) {
            ^
TypeError: a is not a function
    at ////node_modules/@urql/core/src/exchanges/fetch.ts:36:17
    at ////node_modules/wonka/dist/wonka.js:167:16
    at ////node_modules/wonka/dist/wonka.js:482:60
    at ////node_modules/wonka/dist/wonka.js:609:13
    at Array.forEach (<anonymous>)
    at b (////node_modules/wonka/dist/wonka.js:608:13)
    at ////node_modules/wonka/dist/wonka.js:1146:42
    at ////node_modules/wonka/dist/wonka.js:630:31
    at ////node_modules/wonka/dist/wonka.js:180:49
    at h (////node_modules/wonka/dist/wonka.js:163:57)

I saw a similar issue here, however I believe we are using suspense

Any help would be greatly appreciated, as it is causing a bad developer experience

Thanks

Demo Example Throws Error

let example = [1, 2, 3, 4, 5, 6];

Wonka.fromList(example)
  |> Wonka.filter(x => x mod 2 === 0)
  |> Wonka.map(x => x * 2)
  |> Wonka.forEach(x => print_endline(string_of_int(x)));

error:

>>>> Start compiling
[1/1] Building src/data/WonkaTry.cmj

  We've found a bug for you!
  /Users/WonkaTry.re 4:19-36

  2 │
  3 │ Wonka.fromList(example)
  4 │   |> Wonka.filter(x => x mod 2 === 0)
  5 │   |> Wonka.map(x => x * 2)
  6 │   |> Wonka.forEach(x => print_endline(string_of_int(x)));

  This expression should not be a function, the expected type is
  (. 'a) => bool

>>>> Finish compiling(exit: 1)

Wonka.filter sig is:

(
  Js.Internal.fn(
    [ `Arity_1 of 'a14 ],
    bool
  ),
  Wonka_types.sourceT('a),
  Wonka_types.sinkT('a)
) => unit

Any ideas on how to fix this?

Thank you.

Wonka manually polyfills Symbol.observable which can cause interoperability issues with other Observable packages

One of the issues users may run into in the current situation is that if Wonka is loaded after other Observable objects have already been created with the ponyfill method, Wonka will polyfill the Symbol.observable causing packages to no longer recognise previously created observables, breaking interoperability.

How to create a pull stream using `wonka`?

Hi There,

I'm trying to create a stream that iterates down from a very high number. I can't use fromArray as there would be too many entries and I'm intending to cap them later. But effectively I'd like a stream that will only produce x, x-1, x-2 from a high starting point in pull fashion.

The make function for constructing new streams (https://wonka.kitten.sh/api/sources#make) seems to be push only. What's the way to build a pull/lazy stream?

Package license is innaccurate as of 4.0.5

Wonka is listed as MIT but has checked in OCAML dependencies in version v4.0.5 (commit)

This package should be updated to reflect its license as GPL-3 or remove the offending dependency entirely

toObservable supports spec only partially

According to spec the subscribe method either accepts an object with next/error/complete properties or 3 callbacks. The toObservable seems to only support the former variant.

In the ideal world, it wouldn't be a problem. However, when I try to integrate with library that implements spec partially with the second (callbacks) approach, it hurts interoperability.

Unfortunately, I cannot do a ReasonML so I cannot offer much help. Hopefully, it's not a big problem for someone else.

[Question] Ability to use async await for locking async handlers.

I have been using a pipe from the subscriptions and in subscription, I need to execute some async operation that should lock the handling from other subscriptions.

Basically I do not want to have situation where subscriptions will clash.

in Docs we saying:

also similar to IxJS since Wonka streams will run synchronously if an iterable source runs synchronously.

In this case subscriptions are async. Would subscribe handle async and wait or it will be ignored?

Need fromPromise

An essential feature is being able to convert promises to streams.

Missing sourcemaps in distribution?

I'm using urql (3.x) with create-react-app (5.x) and it appears the dev server is unable to find proper sourcemaps for wonka in particular:

WARNING in ./node_modules/wonka/dist/wonka.mjs
Module Warning (from ./node_modules/source-map-loader/dist/cjs.js):
Failed to parse source map from '[path_to_project]/node_modules/src/callbag.ts' file: Error: ENOENT: no such file or directory, open '[path_to_project]/node_modules/src/callbag.ts'

This means debugging urql/wonka becomes more difficult, and it also makes a lot of noise for the React devserver logs that can make other issues easier to miss.

Any ideas what might be going on? Does wonka intentionally not ship with sourcemaps currently?

Indicate that `this` is not used, or convert functions to arrow functions

A few of the functions in Wonka triggers the @typescript-eslint/unbound-method lint rule when used. See https://github.com/typescript-eslint/typescript-eslint/blob/main/packages/eslint-plugin/docs/rules/unbound-method.md.

For example, the makeSubject example from the documentation:

import { makeSubject } from 'wonka'
const subject = Wonka.makeSubject();
const { source, next, complete } = subject;

Changing export function makeSubject<T>(): Subject<T> { to export function makeSubject<T>(this: void): Subject<T> { should be enough to indicate that this is not used in these functions. Or it could just be an arrow function instead in which case we do not need to indicate that.

Subject

Before I open new PR, I would like to know if this is a right direction. Could you please take a look at the following Subject implementation (inspired by callbag-subject) and let me know what do you think.

re

module Subject = {
  type subjectStateT('a) = {
    sinks: Belt.MutableMap.Int.t(signalT('a) => unit),
    mutable idCounter: int,
  };

  let make = () => {
    let state = {sinks: Belt.MutableMap.Int.make(), idCounter: 0};

    signal =>
      switch (signal) {
      | Start(_) => (
          sink => {
            let id = state.idCounter;
            Belt.MutableMap.Int.set(state.sinks, id, sink);
            state.idCounter = state.idCounter |> succ;

            sink(
              Start(
                signal =>
                  switch (signal) {
                  | End => Belt.MutableMap.Int.remove(state.sinks, id)
                  | _ => ()
                  },
              ),
            );
          }
        )
      | _ =>
        Belt.MutableMap.Int.forEachU(state.sinks, (. _, sink) =>
          sink(signal)
        );
        (_ => ());
      };
  };

  let toStream = subject => subject(Start(_ => ()));

  let next = (value, subject) => subject(Push(value), _ => ());

  let complete = subject => subject(End, _ => ());
};

rei

module Subject: {
  let make: (unit, signalT('a), signalT('a) => unit) => unit;
  let toStream: (signalT('a) => 'b) => 'b;
  let next: ('a, (signalT('a), 'b => unit) => 'c) => 'c;
  let complete: ((signalT('a), 'b => unit) => 'c) => 'c;
};

Example:

let subject = Wonka.Subject.make();

Wonka.(
  subject
  |> Subject.toStream
  |> map(succ)
  |> forEach(Js.log)
);

Wonka.Subject.(subject |> next(6));

Add makeReplaySubject source

This should act as a subject, with the addition of emitting immediately old values to new subscribers, in FIFO order.

The api could look like:

let makeReplaySubject: (bufferSize: int) => subjectT('a) = ....;

// example:
let subject = makeReplaySubject(5);

I'm currently working on a PR, @kitten please tell me if you think this could be useful!

Observer.next from make doesn't accept boolean:false

After playing around with wonka and trying to make a source with make, if you pass false as the next value you get uncaught error.

Should I be able to pass a boolean value as the next value?

How to use combine?

Two arguments in example.

But

https://github.com/kitten/wonka/blob/f26cc6c1cbcbf4f3240a121575492c624f1e1980/src/operators/wonka_operator_combine.d.ts#L3

Are the source docs up-to-date?

Using the docs for makeSubject I did:

let subject = Wonka.makeSubject();
let (source, next, complete) = subject;

and get the error:

This has type:
    Wonka_types.subjectT('a)
  But somewhere wanted:
    ('b, 'c, 'd)

Similarly the example for make doesn't seem to compile (the error is too inscrutable for me to figure out why, but seems to be a type issue)

sourcing `./src/Wonka.bs.js`

I ran wonka from a snowpack generated project and got this random error that ./src/Wonka.bs.js could not be sourced.

Reproduction: https://github.com/idkjs/wonka-snowpack

Any idea what is going on here?

Thank you, sir.

Having just seen #97, I am on node v14.15.4

> sw_vers
ProductName:	macOS
ProductVersion:	11.2
BuildVersion:	20D64
> node -v
v14.15.4
>

Error DeprecationWarning: Use of deprecated folder mapping "./" in the "exports" field module resolution of the package

The following error occurred while coding.

[DEP0148] DeprecationWarning: Use of deprecated folder mapping "./" in the "exports" field module resolution of the package at C:\src\node_modules\wonka\package.json imported from C:\src\node_modules.
Update this package.json to use a subpath pattern like "./*".
(Use `node --trace-deprecation ...` to show where the warning was created)

It seems that nodejs version 16 requires /*.
Can I ask you to correct this one?

Best regards.

Add backpressure operators

There are certain cases where sinks don't consume push signals as fast as they're sent. This can often happen with an asynchronous source that is fed through a delay or an asynchronous sink.

In such cases currently you can only use debounce or throttle (maybe even buffer) but it'd be nice to have actual backpressure operators.

Such operators would not let new values through while no Pull event has come back.

Buffer operator

Hi! I'm trying to implement request batching in urql. I'm not at all familiar with rxjs-style APIs, but as far as I can tell, this requires a buffer and/or bufferTime operator in wonka. (I'm assuming that you can define bufferTime in terms of buffer.)

If I'm right about all of that, is this an enhancement you're open to?

(For now I've implemented one local to my project in JS. That was somewhat painful due to the generated JS for some of the types not being very easy to use. Flow is not impressed with attempts to produce objects satisfying this type:

export type Signal<A> =
  | ({
      tag: 0
    } & [(talkback: Talkback) => void])
  | ({
      tag: 1
    } & [A])
  | 0;

)

4.0.5 Flow Types are broken

Sampling:

Error ----------------------------------------------- node_modules/wonka/dist/types/src/Wonka_operators.gen.js.flow:3:10

Cannot import the type operatorT as a value. Use import type instead.

     1| // @flow
     2|
     3| import { operatorT as Wonka_types_operatorT } from "./Wonka_types.gen";
     4| import { sourceT as Wonka_types_sourceT } from "./Wonka_types.gen";
     5| declare export var buffer: <a, b>(
     6|   _1: Wonka_types_sourceT<a>


Error ----------------------------------------------- node_modules/wonka/dist/types/src/Wonka_operators.gen.js.flow:4:10

Cannot import the type sourceT as a value. Use import type instead.

     1| // @flow
     2|
     3| import { operatorT as Wonka_types_operatorT } from "./Wonka_types.gen";
     4| import { sourceT as Wonka_types_sourceT } from "./Wonka_types.gen";
     5| declare export var buffer: <a, b>(
     6|   _1: Wonka_types_sourceT<a>
     7| ) => Wonka_types_operatorT<b, b[]>;


Error --------------------------------------------------- node_modules/wonka/dist/types/src/Wonka_sinks.gen.js.flow:3:10

Cannot import the type sourceT as a value. Use import type instead.

     1| // @flow
     2|
     3| import { sourceT as Wonka_types_sourceT } from "./Wonka_types.gen";
     4| import { subscriptionT as Wonka_types_subscriptionT } from "./Wonka_types.gen";
     5| export type subscribeConsumerT<a> = (
     6|   _1: Wonka_types_sourceT<a>

Looks like it's due to this: joarwilk/flowgen#74

I'll be working around this with a local fork of the generated typedefs, but I figured you'd want to know.

toPromise doesn't reject when erroring

toPromise will never reject / call catch().

It makes it impossible to handle locally errors.

try {
  await client.mutation(mutation).toPromise()
}
catch(e) {
  console.warn("error");
} 

won't work.

`switchMap` example output not as expected in docs

Is the output,000, correct or are the docs correct? Thanks.

> node src/SwitchMap.bs.js
## switchMap

`switchMap` allows you to map values of an outer source to an inner source.
The inner source's values will be emitted on the returned output source. If
a new inner source is returned, because the outer source emitted a new value
before the previous inner source completed, the inner source is closed and unsubscribed
from.

This is similar to `concatMap` but instead of waiting for the last inner source to complete
before emitting values from the next, `switchMap` just cancels the previous inner source.

```reason
Wonka.interval(50)
  |> Wonka.switchMap((. _value) =>
    Wonka.interval(40))
  |> Wonka.take(3)
  |> Wonka.subscribe((. x) => print_int(x));
/* Prints 1 2 3 to the console. */
/* The inner interval is cancelled after its first value every time */
000⏎                                         > 

Add distinct operator

This should be able to filter out values when they're not distinct. Instead of just using reference equality there could be an isDistinct predicate function input.

So the default usage may be: distinct((a, b) => a === b)

How do you deal with errors?

Looking through the readme, examples, and code, I don't see how to deal with errors whether explicit or whether handled with something like Result.

Obviously if one of the operators or sources returns an error, then it should short circuit the processing in the pipeline.

Since promises, observables, and callbags all can have errors, how are they dealt with?

Using `Wonka.fromObservable` in Reason API

Lets say you have and observable graphql subscription which in JS looks like:

	useEffect(() => {
		const subscription = API.graphql(graphqlOperation(onCreateMessage)).subscribe({
			next: (event) => {
				if (event) {
					console.log('Subscription: ' + JSON.stringify(event.value.data, null, 2));
					console.log('EVENT: ' + JSON.stringify(event, null, 2));
					setDisplay(true);
					let message = event.value.data.onCreateMessage.message;
					setMessage(message);
				}
			}
		});

The subscribe function is return a zen-observable which i believe is the same one returned by Wonka. See: https://github.com/aws-amplify/amplify-js/blob/86597db2b0704d7aff5b612557536142b82e1731/packages/api/src/API.ts#L16

How would I access the return value using Wonka.fromObservable?

This is what I have tried:

type t;
[@bs.module "@aws-amplify/api"] external api: t = "default";

[@bs.send]
external _graphqlSubWonka:
  (t, Types.graphqlOperation) => Wonka.observableT('a) =
  "graphql";

graphql is binding to API.ts#L352

In useEffect i have got:

  React.useEffect(() => {
    let subRequest = Graphql.OnCreateMessage.make();
    let graphqlOperation: Types.graphqlOperation = {
      query: subRequest##query,
      variables: subRequest##variables,
	};
    API.subWithWonka(graphqlOperation)
    |> Wonka.fromObservable((. result) => Js.log(result));
    None;
  });

This yield the following error which I do not understand how to read:

Error: This expression has type (. 'a) => unit
       but an expression was expected of type
         Wonka.observableT('b) = Wonka_observable.observableT('b)

I have tried a bunch of other things but just give this simple example to get the conversation started. Thank you for sharing your project.

.rei files in the readme are missing

https://github.com/kitten/wonka/blame/master/README.md#L121-L122

Latest operator

Is there a way to create a "latest" operator? This would be an operator that returns only the most recent value emitted from the source (similar to buffer but returning only the last element of the buffer).

The reason I need this is because I'm trying to reduce the number of times something can be called (similar to throttle). So if I call the function x(), the system should wait some number of ms, and additional calls to x() don't do anything in that time period. After that x() is called.

None of the existing operators seems to match this use case and I'm not sure how to construct it from the current operators. I think this would be the equivalent of a .latest().delay(1000)

(scan doesn't work because it requires an initial value and the initial value isn't available and the type isn't nullable)

Thanks!

convert ECMAScript Observable to RXJS Observable

Is it possible to convert the observable to an rxjs observable?

This would be the desired behavior from an urql subscription:

const sub =  pipe(
  this.client.subscription(q),
  toObservable
);

sub.subscribe((r: any) {
  console.log(r);
});

This seems like a lot more complicated to write every time:

.subscribe({
  next(r: any) {
    console.log(r);
  },
  error(e: any) {
    console.log('Error: ', e);
  },
  complete() { }
});

Also, in Angular, if I use the observable in the template, I get this error:

ERROR TypeError: Cannot read property 'bind' of undefined

So, I hacked it using this, although it seems like there should be an easier way without this hack:

test: BehaviorSubject<any> = new BehaviorSubject<any>([]);;
sub.subscribe({
      next(r: any) {
        _this.test.next(r);
        console.log(r);
      },
      error(e: any) { },
      complete() { }
    });

Then I use test in my template...

Is this the only way?

Newer versions of node can't use Wonka

Some of the newest versions of NPM require the package.json's exports to include wonka.bs.js . Sorry, I'm not providing a lot of detail right now, just wanted to write this down before I forgot.

Inconsistent behaviour when adding buffer() inside concatMap()

First of all my apologies, this might be a stupid user error problem or a real bug - I am new to using Wonka and cannot yet tell which one it is. :)

I am implementing my own bufferCount() operator by concatMap and nested buffer inside. At some point I wondered why the buffers always emit with size of 1 and resorted back to the original buffering example. I suspect there is a bug somewhere or a concept I don't understand, so here is a simplified example - this time with interval timers instead of a counter.

Please help me figure out what am I missing. Alternatively, I would appreciate a bufferCount() implementation as that is my target anyway. :)

Wonka example on buffering works alright:

pipe(
  interval(100),
  buffer(interval(1000))
  // prints [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], [11, 12, ...], etc...
  subscribe(console.log)
)

However, when nested inside concatMap, the arrays emit one item at time

pipe(
  interval(100),
  concatMap(value =>
    pipe(
      fromValue(value),
      buffer(interval(1000))
    )
  ),
  // prints [1], [2], [3] etc...
  subscribe(console.log)
)

Nice Commented example Wonka usage

see https://github.com/nicksenger/senger-io.

Especially Epics.re and UseEpicReducer.re.

Thanks to @nicksenger

Sample Example output not what is expected

let _= Wonka.interval(10)
  |> Wonka.sample(Wonka.interval(100))
  |> Wonka.take(2)
  |> Wonka.subscribe((. x) => print_int(x));

> node src/Sample.bs.js
## sample

`sample` emits the previously emitted value from an outer source every time
an inner source (notifier) emits.

In combination with `interval` it can be used to get values from a noisy source
more regularly.

/* Prints 10 20 to the console. */
715⏎  

Also, are there any tests for the reasonml code?

`dist/wonka.d.ts` failing typecheck on TS 4.9.3+

CI is failing for a package I maintain that uses wonka via urql, and typechecks with skipLibCheck: false and isolatedModules: true. I am now getting the following errors after the release of 6.2.0 and 6.2.1:

claw ~/C/scratch-winka ➜  npx tsc
../node_modules/wonka/dist/wonka.d.ts:1296:5 - error TS1169: A computed property name in an interface must refer to an expression whose type is a literal type or a 'unique symbol' type.

1296     [Symbol.observable]?(): Observable<T>;
         ~~~~~~~~~~~~~~~~~~~

../node_modules/wonka/dist/wonka.d.ts:1296:13 - error TS2339: Property 'observable' does not exist on type 'SymbolConstructor'.

1296     [Symbol.observable]?(): Observable<T>;
                 ~~~~~~~~~~

../node_modules/wonka/dist/wonka.d.ts:1325:5 - error TS1169: A computed property name in an interface must refer to an expression whose type is a literal type or a 'unique symbol' type.

1325     [Symbol.observable](): Observable<T>;
         ~~~~~~~~~~~~~~~~~~~

../node_modules/wonka/dist/wonka.d.ts:1325:13 - error TS2339: Property 'observable' does not exist on type 'SymbolConstructor'.

1325     [Symbol.observable](): Observable<T>;
                 ~~~~~~~~~~

../node_modules/wonka/dist/wonka.d.ts:1428:44 - error TS2748: Cannot access ambient const enums when the '--isolatedModules' flag is provided.

1428 export { Observer, Operator, Push, Signal, SignalKind, Sink, Source, Start, Subject, Subscription, Tag, TalkbackFn, TalkbackKind, TeardownFn, TypeOfSource, buffer, combine, concat, concatAll, concatMap, debounce, delay, empty, filter, mergeAll as flatten, forEach, fromArray, fromAsyncIterable, fromCallbag, fromDomEvent, fromIterable, fromObservable, fromPromise, fromValue, interval, lazy, make, makeSubject, map, merge, mergeAll, mergeMap, never, onEnd, onPush, onStart, pipe, publish, sample, scan, share, skip, skipUntil, skipWhile, subscribe, switchAll, switchMap, take, takeLast, takeUntil, takeWhile, onPush as tap, throttle, toArray, toAsyncIterable, toCallbag, toObservable, toPromise, zip };
                                                ~~~~~~~~~~

../node_modules/wonka/dist/wonka.d.ts:1428:117 - error TS2748: Cannot access ambient const enums when the '--isolatedModules' flag is provided.

1428 export { Observer, Operator, Push, Signal, SignalKind, Sink, Source, Start, Subject, Subscription, Tag, TalkbackFn, TalkbackKind, TeardownFn, TypeOfSource, buffer, combine, concat, concatAll, concatMap, debounce, delay, empty, filter, mergeAll as flatten, forEach, fromArray, fromAsyncIterable, fromCallbag, fromDomEvent, fromIterable, fromObservable, fromPromise, fromValue, interval, lazy, make, makeSubject, map, merge, mergeAll, mergeMap, never, onEnd, onPush, onStart, pipe, publish, sample, scan, share, skip, skipUntil, skipWhile, subscribe, switchAll, switchMap, take, takeLast, takeUntil, takeWhile, onPush as tap, throttle, toArray, toAsyncIterable, toCallbag, toObservable, toPromise, zip };
                                                                                                                         ~~~~~~~~~~~~


Found 6 errors in the same file, starting at: ../node_modules/wonka/dist/wonka.d.ts:1296

To reproduce, in a new folder:

npm install wonka typescript

Add the following to tsconfig.json:

{
  "compilerOptions": {
    "lib": ["dom", "ES2018"],
    "target": "es2016",                                  
    "module": "commonjs",                                
    "esModuleInterop": true,                             
    "forceConsistentCasingInFileNames": true,            
    "strict": true,                                      
    "skipLibCheck": false,
    "isolatedModules": true,                                 
  }
}

Add the following to index.ts:

import wonka from "wonka";

And then run npx tsc --noEmit.

How to avoid intermediary values when combine uses the same input sources twice?

Is it possible to avoid glitches with synchronous receiving data from a few sources?

import w from 'wonka'

const log = []
const s = w.makeSubject()
w.pipe(
  w.combine(s.source, s.source),
  w.subscribe(v => log.push(v)),
)

s.next(0)
s.next(1)

console.assert(
  JSON.stringify(log) === JSON.stringify([ [0, 0], [1, 1] ]) // false
)

In example above the log will be [[0,0],[1,0],[1,1]], where [1,0] is redundant data (glitch)

Question about onStart execution order

In the following example I was expecting "onStart" callback to be invoked first, but noticed it was run last:

https://codepen.io/ejez/pen/NWjWBoJ

const src = fromArray([0, 1]);

const { unsubscribe } = pipe(
  src,

  onStart(() => console.log("onStart...")),

  onPush(() => console.log("onPush...")),

  onEnd(() => console.log("onEnd...")),

  subscribe((result) => console.log(result))
);

result:

"onPush..."
0
"onPush..."
1
"onEnd..."
"onStart..."

From the docs:

When the stream starts then the sink is called with Start, Then for every incoming, new value it’s called with Push('a), and when the stream ends it’s finally called with End.

onStart
Run a callback when the Start signal is sent to the sink by the source.

Many thanks.

New sources and operators? (roadmap)

hello there! 👋 I have been waiting for the next Wonka release for a long time, and I'm happy to see v6!

Although, I'm a bit concerned about missing new source creators and operators in this specific release. Functions like: combineLatest, forkJoin, makeBehaviorSubject, startWith, endWith, pairwise, distinct, distinctUntilChanged etc. would be really great addition to Wonka, is this something on your roadmap? (I'd be happy to help with implementing a few if you need help)

Refactor to uncurry as much as possible by default

b is not a function

I'm using urql in a React app, which pulls in Wonka as a dependency, and I'm getting the following error during SSR on any of my pages that does a GraphQL query:

TypeError: b is not a function
at d (webpack-internal:////node_modules/wonka/dist/wonka.mjs:125:203)
at eval (webpack-internal:////node_modules/wonka/dist/wonka.mjs:700:9)
at eval (webpack-internal:////node_modules/urql/dist/urql.es.js:258:15)
at eval (webpack-internal:////node_modules/wonka/dist/wonka.mjs:692:14)
at eval (webpack-internal:////node_modules/wonka/dist/wonka.mjs:136:144)
at _0 (webpack-internal:////node_modules/wonka/dist/wonka.mjs:72:55)
at _0 (webpack-internal:////node_modules/wonka/dist/wonka.mjs:146:32)
at d (webpack-internal:////node_modules/wonka/dist/wonka.mjs:247:140)
at eval (webpack-internal:////node_modules/wonka/dist/wonka.mjs:138:7)
at eval (webpack-internal:////node_modules/wonka/dist/wonka.mjs:257:143)

I tried going into wonka.mjs as well as urql.es.js but they're all optimized and I can't make heads or tails out of the line numbers. I know that it wasn't always throwing this error, and I'm not sure what I changed to break it.

The only thing that I can think of is that I recently installed the reason-urql bindings and the graphql-ppx for ReScript as I'm migrating my app over to ReasonReact and ReScript. However, I'm not currently using either the bindings or the ppx.

Can anyone help me out? I'm sorry, I know that it's not a lot to go on. This is probably more of a debugging question than it is an issue with the library.

`Wonka.throttle` example emits `0 5` instead of `0 6`

> node src/Throttle.bs.js 
## throttle

`throttle` emits values of a source, but after each value it will omit all values for
the given amount of milliseconds. It enforces a time of silence after each value it
receives and skips values while the silence is still ongoing.

This is very similar to `debounce` but instead of waiting for leading time before a
value it waits for trailing time after a value.

> _Note:_ This operator is only available in JavaScript environments, and will be excluded
> when compiling natively.

```reason
Wonka.interval(10)
  |> Wonka.throttle((. _x) => 50)
  |> Wonka.take(2)
  |> Wonka.subscribe((. x) => print_int(x));
/* Outputs 0 6 to the console. */

05⏎ >

Package not found: resolving package wonka in /

Package not found: resolving package wonka in /
I get this error when running bsb -make-world

GPL license in opam/conf-m4

Currently being flagged by fossa.com as a potential GPL issue

esy.lock/opam/conf-m4.1

Can anyone confirm that this is an actual issue or do we sidestep it because it's only used in the build process or similar?

Thanks.

`Wonka.share` type contains type variables that can't be generalized: Wonka_types.sourceT('_weak1)

Running the docs example, I get:

let source = Wonka.never
  |> Wonka.onStart((. ) => print_endline("start"))
  |> Wonka.share;

Wonka.publish(source);
Wonka.publish(source);

Output

 let source = Wonka.never
  21 │   |> Wonka.onStart((. ) => print_endline("start"))
  22 │   |> Wonka.share;
  
  This expression's type contains type variables that can't be generalized:
  Wonka_types.sourceT('_weak1)
  
  This happens when the type system senses there's a mutation/side-effect,
  in combination with a polymorphic value.
  Using or annotating that value usually solves it.
  
FAILED: cannot make progress due to previous errors.
>>>> Finish compiling(exit: 1)

How would you annotate this function to get it to work? Thank you!

Combine N sources

I'm not sure about what's the best way to combine N sources with the combine operator.
As described in the docs, the combine operator supports only 2 sources right now.
Any idea on how to create a operator that combines a list/array of sources instead?

pairwise operator

Is have pairwise operator in wonka?

Runtime error when used with BuckleScript 8.1.0

The new BS version seems to break, at runtime, the wonka take function:

The generated code for the take function looks something like this:

if (state.taken < max) {
  if (signal) {
    state.taken = max;
    return state.talkback(/* Close */1); // Breaks here
  } else {
    return state.talkback(/* Pull */0);
  }
}

When logging, the stage.talkback propery is indeed undefined 😞

Docs?

Are there any docs on this library? I see that it's used in urql and I'm trying to understand it but it's going way over my head 😛