Skip to content
Permalink
Branch: master
Find file Copy path
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
232 lines (173 sloc) 5.69 KB
title order
Sinks
2

A sink in Wonka expects to be delivered data. A sink communicates with a source via the "talkback" function provided by the source. Wonka has the following sink operators.

subscribe

subscribe accepts a callback function to execute when data is received from the source, in addition to the source itself.

Wonka.fromArray([|1, 2, 3|])
  |> Wonka.subscribe((. x) => print_int(x));
/* Prints 123 to the console. */
import { pipe, fromArray, subscribe } from 'wonka';

pipe(
  fromArray([1, 2, 3]),
  subscribe((x) => console.log(x))
); // Prints 123 to the console.

subscribe also returns a "subscription" type, which can be used to unsubscribe from the source. This allows you to cancel a source and stop receiving new incoming values.

let subscription = source
  |> Wonka.subscribe((. x) => print_int(x));

subscription.unsubscribe();
import { pipe, subscribe } from 'wonka';

const [unsubscribe] = pipe(
  source,
  subscribe((x) => console.log(x));
);

unsubscribe();

forEach

forEach works the same as subscribe but doesn't return a subscription. It will just call the passed callback for each incoming value.

Wonka.fromArray([|1, 2, 3|])
  |> Wonka.forEach((. x) => print_int(x));
/* Returns unit; Prints 123 to the console. */
import { pipe, fromArray, forEach } from 'wonka';

pipe(
  fromArray([1, 2, 3]),
  forEach((x) => console.log(x))
); // Returns undefined; Prints 123 to the console.

publish

publish subscribes to a source, like subscribe does, but doesn't accept a callback function. It's useful for side-effects, where the values are already being used as part of the stream itself.

In this example we're using onPush to pass a callback to react to incoming values instead.

Wonka.fromArray([|1, 2, 3|])
  |> Wonka.onPush((. x) => print_int(x))
  |> Wonka.publish;
/* Prints 123 to the console. */
import { pipe, fromArray, onPush, publish } from 'wonka';

pipe(
  fromArray([1, 2, 3]),
  onPush((x) => console.log(x)),
  publish
); // Prints 123 to the console.

toArray

toArray returns an array, which contains all values from a pull source. This sink is primarily intended for synchronous pull streams. Passing it an asynchronous push streams may result in an empty array being returned.

If you're passing an asynchronous push stream toArray will cancel it before it returns an array.

Note: If you're using this sink, make sure that your input source streams the values you're collecting partly or fully synchronously.

Wonka.fromArray([|1, 2, 3|])
  |> Wonka.map((. x) => x * 2)
  |> Wonka.toArray
/* Returns [|2, 4, 6|] */
import { pipe, fromArray, map, toArray } from 'wonka';

pipe(
  fromArray([1, 2, 3]),
  map(x => x * 2),
  toArray
); // Returns [2, 4, 6]

toPromise

toPromise returns a promise, which resolves on the last value of a source.

Note: This source is only available in JavaScript environments, and will be excluded when compiling natively.

Wonka.fromArray([|1, 2, 3|])
  |> Wonka.toPromise
  |> Js.Promise.then_(x => {
    print_int(x);
    Js.Promise.resolve(())
  })
/* Prints 3 to the console. */
import { pipe, fromArray, toPromise } from 'wonka';

const promise = pipe(
  fromArray([1, 2, 3]),
  toPromise,
);

promise.then(x => console.log(x));
// Prints 3 to the console.

If you have a source that doesn't complete and are looking to resolve on the first value instead of the last, you may have to apply take(1) to your source.

toObservable

toObservable returns a spec-compliant JS Observable, which emits the same values as a source.

As per the specification, the Observable is annotated using Symbol.observable.

Note: This sink is only available in JavaScript environments, and will be excluded when compiling natively.

let observable = Wonka.fromArray([|1, 2, 3|])
  |> Wonka.toObservable;

observable##subscribe([@bs] {
  as _;
  pub next = value => print_int(value);
  pub complete = () => ();
  pub error = _ => ();
}); /* Prints 1 2 3 to the console. */
import { pipe, fromArray, toObservable } from 'wonka';

const observable = pipe(
  fromArray([1, 2, 3]),
  toObservable,
);

observable.subscribe({
  next: value => console.log(value),
  complete: () => {},
  error: () => {},
}); // Prints 1 2 3 to the console.

toCallbag

toCallbag returns a spec-compliant JS Callbag, which emits the same signals as a Wonka source.

Since Wonka's sources are very similar to callbags and only diverge from the specification minimally, Callbags map to Wonka's sources very closely and toCallbag only creates a thin wrapper which is mostly concerned with converting between the type signatures.

Note: This sink is only available in JavaScript environments, and will be excluded when compiling natively.

/* This example uses the callbag-iterate package for illustrative purposes */
[@bs.module] external callbagIterate:
  (. ('a => unit)) => (. Wonka.callbagT('a)) => unit = "callbag-iterate";

let callbag = Wonka.fromArray([|1, 2, 3|])
  |> Wonka.toCallbag;

callbagIterate(. value => {
  print_int(value);
})(. callbag); /* Prints 1 2 3 to the console. */
import { pipe, fromArray, toCallbag } from 'wonka';

// This example uses the callbag-iterate package for illustrative purposes
import callbagIterate from 'callbag-iterate';

const callbag = pipe(
  fromArray([1, 2, 3]),
  toCallbag,
);

callbagIterate(value => console.log(value))(callbag);
// Prints 1 2 3 to the console.
You can’t perform that action at this time.