Permalink
Branch: master
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
1558 lines (923 sloc) 40 KB

API

@most/types

Time

All time-related types use units defined by a :ref:`Clock`. The default :ref:`Scheduler` :ref:`Clock` uses milliseconds as its units: :ref:`Time`, :ref:`Delay <Delay-type>`, :ref:`Period`, and :ref:`Offset` will all be millisecond values.

type Time = number

Time is a monotonic number. It represents the current time according to a :ref:`Clock`. When using a default :ref:`Scheduler <newDefaultScheduler>`, the units will be milliseconds.

Delay

type Delay = number

A Delay represents a duration from "now". When using a default :ref:`Scheduler <newDefaultScheduler>`, the units will be milliseconds.

Period

type Period = number

A Period represents a regular interval. When using a default :ref:`Scheduler <newDefaultScheduler>`, the units will be milliseconds.

Offset

type Offset = number

An Offset represents the relationship of one :ref:`Clock` to another. When using a default :ref:`Scheduler <newDefaultScheduler>`, the units will be milliseconds.

NOTE: Typically, you will not need to be concerned with the :ref:`Offset` type.

Stream

type Stream a = {
  run :: Sink a -> Scheduler -> Disposable
}

A Stream represents a view of events over time. Its run method arranges events to be propagated to the provided :ref:`Sink` in the future. Each Stream has a local clock, defined by the provided :ref:`Scheduler`, which has methods for knowing the current time and scheduling future :ref:`Tasks <Task>`.

A Stream may be simple, like :ref:`now`, or may do sophisticated things such as :ref:`combining <combine>` multiple Stream s or deal with higher-order Stream s.

A Stream may act as an event producer, such as a Stream that produces DOM events. A producer Stream must never produce an event in the same call stack as its run method is called. It must begin producing items asynchronously. In some cases, this comes for free, such as DOM events. In other cases, it must be done explicitly using the provided :ref:`Scheduler` to schedule asynchronous :ref:`Tasks <Task>`.

Sink

type Sink a = {
  event :: Time -> a -> void
  error :: Time -> Error -> void
  end :: Time -> void
}

A Sink receives events—typically it does something with them, such as transforming or filtering them—and then propagates them to another Sink.

Typically, a combinator will be implemented as a :ref:`Stream` and a Sink. The :ref:`Stream` is usually stateless/immutable and creates a new Sink for each new observer. In most cases, the relationship of a :ref:`Stream` to Sink is 1-many.

Disposable

type Disposable = {
  dispose:: () -> void
}

A Disposable represents a resource that must be disposed of (or released), such as a DOM event listener.

Scheduler

type Scheduler = {
  currentTime :: () -> Time
  scheduleTask :: Offset -> Delay -> Period -> Task -> ScheduledTask
  relative :: Offset -> Scheduler
  cancel :: ScheduledTask -> void
  -- deprecated
  cancelAll :: (ScheduledTask -> boolean) -> void
}

A Scheduler provides the central notion of time for the :ref:`Streams <Stream>` in an application.

An application will typically create a single "root" Scheduler so that all :ref:`Streams <Stream>` share the same underlying time.

Clock

type Clock = {
  now :: () -> Time
}

A Clock represents a source of the current time. The default :ref:`Clock` uses milliseconds as its units: :ref:`Time`, :ref:`Delay-type`, :ref:`Period`, and :ref:`Offset` will all be millisecond values.

Timer

type Handle = any -- intentionally opaque handle

type Timer = {
  now :: () -> Time,
  setTimer :: (() -> any) -> Delay -> Handle,
  clearTimer :: Handle -> void
}

A Timer abstracts platform time, typically relying on a :ref:`Clock`, and timer scheduling, typically using setTimeout.

Timeline

type TaskRunner = (ScheduledTask) -> any

type Timeline = {
  add :: ScheduledTask -> void,
  remove :: ScheduledTask -> boolean,
  -- deprecated
  removeAll :: (ScheduledTask) -> boolean) -> void,
  isEmpty :: () -> boolean,
  nextArrival :: () -> Time,
  runTasks :: Time -> TaskRunner -> void
}

A Timeline represents a set of :ref:`ScheduledTasks <ScheduledTask>` to be executed at particular times.

Task

type Task = Disposable & {
  run :: Time -> void,
  error:: Time -> Error -> void
}

A Task is any unit of work that can be scheduled for execution with a :ref:`Scheduler`.

ScheduledTask

type ScheduledTask = Disposable & {
  task :: Task,
  run :: () -> void,
  error :: Error -> void
}

A ScheduledTask represents a :ref:`Task` which has been scheduled in a particular :ref:`Scheduler`. A ScheduledTask's dispose method will cancel the :ref:`Task` with the :ref:`Scheduler` with which it was scheduled.

@most/core

Running

runEffects

runEffects :: Stream a -> Scheduler -> Promise void

Activate an event :ref:`Stream` and consume all its events.

run

Attention!

@most/core encourages a declarative approach. Combinators like :ref:`until` allow you to declare which events you're interested in, and @most/core will manage acquiring and disposing resources automatically. run is intended for use cases that cannot be handled declaratively, such as at integration points with other projects whose APIs may force an imperative approach.

run :: Sink a -> Scheduler -> Stream a -> Disposable

Run a :ref:`Stream`, sending all events to the provided :ref:`Sink`. The Stream's :ref:`Time` values come from the provided :ref:`Scheduler`. Returns a :ref:`Disposable` that can be used to dispose underlying resources imperatively.

Declarative combinators like :ref:`until` still manage resources automatically when using run. The returned :ref:`Disposable` simply provides an additional way to trigger disposal manually.

Construction

empty

empty :: () -> Stream *

Create a :ref:`Stream` containing no events and ends immediately.

empty(): |

never

never :: () -> Stream *

Create a :ref:`Stream` containing no events and never ends.

never(): ---->

now

now :: a -> Stream a

Create a :ref:`Stream` containing a single event at time 0.

now(x): x|

at

at :: Time -> a -> Stream a

Create a :ref:`Stream` containing a single event at a specific time.

at(3, x): --x|

periodic

periodic :: Period -> Stream void

Create an infinite :ref:`Stream` containing events that occur at a specified :ref:`Period`. The first event occurs at time 0, and the event values are undefined.

periodic(3): x--x--x--x-->

throwError

throwError :: Error -> Stream void

Create a :ref:`Stream` that fails with the provided Error at time 0. This can be useful for functions that need to return a :ref:`Stream` and also need to propagate an error.

throwError(X): X

Extending

startWith

startWith :: a -> Stream a -> Stream a

Prepend an event at time 0.

stream:               --a-b-c-d->
startWith(x, stream): x-a-b-c-d->

Note that startWith does not delay other events. If stream already contains an event at time 0, then startWith simply adds another event at time 0—the two will be simultaneous, but ordered. For example:

stream:                a-b-c-d->
startWith(x, stream): xa-b-c-d->

Both x and a occur at time 0, but x will be observed before a.

continueWith

continueWith :: (() -> Stream a) -> Stream a -> Stream a

Replace the end of a :ref:`Stream` with another :ref:`Stream`.

stream:                  -a-b-c-d|
f():                                   -1-2-3-4-5->
continueWith(f, stream): -a-b-c-d-1-2-3-4-5->

When stream ends, f will be called and must return a :ref:`Stream`.

Transformation

map

map :: (a -> b) -> Stream a -> Stream b

Apply a function to each event value.

stream:         -a-b-c-d->
map(f, stream): -f(a)-f(b)-f(c)-f(d)->
map(x => x + 1, stream)

constant

constant :: a -> Stream * -> Stream a

Replace each event value with x.

stream:              -a-b-c-d->
constant(x, stream): -x-x-x-x->
constant('tick', periodic(1000))

tap

tap :: (a -> *) -> Stream a -> Stream a

Perform a side effect for each event in a :ref:`Stream`.

stream:         -a-b-c-d->
tap(f, stream): -a-b-c-d->

For each event in stream, f is called, but the value of its result is ignored. If f fails (i.e., throws an error), then the returned :ref:`Stream` will also fail. The :ref:`Stream` returned by tap will contain the same events as the original :ref:`Stream`.

ap

ap :: Stream (a -> b) -> Stream a -> Stream b

Apply the latest function in a :ref:`Stream` of functions to the latest value of another :ref:`Stream`.

streamOfFunctions:              --f-----------g---------h--------->
stream:                         -a-------b---------c---------d---->
ap(streamOfFunctions, stream): --f(a)---f(b)-g(b)-g(c)-h(c)-h(d)->

In effect, ap applies a time-varying function to a time-varying value.

scan

scan :: (b -> a -> b) -> b -> Stream a -> Stream b

Incrementally accumulate results, starting with the provided initial value.

stream:                           -1-2-3->
scan((x, y) => x + y, 0, stream): 01-3-6->

loop

loop :: (b -> a -> { seed :: b, value :: c }) -> b -> Stream a -> Stream c

Accumulate results using a feedback loop that emits one value and feeds back another to be used in the next iteration.

It allows you to maintain and update a "state" (a.k.a. feedback, a.k.a. seed for the next iteration) while emitting a different value. In contrast, :ref:`scan` feeds back and produces the same value.

// Average an array of values.
const average = values =>
      values.reduce((sum, x) => sum + x, 0) / values.length

const stream = // ...

// Emit the simple (i.e., windowed) moving average of the 10 most recent values.
loop((values, x) => {
      values.push(x)
      values = values.slice(-10) // Keep up to 10 most recent
      const avg = average(values)

      // Return { seed, value } pair.
      // seed will feed back into next iteration.
      // value will be propagated.
      return { seed: values, value: avg }
}, [], stream)

zipItems

zipItems :: ((a, b) -> c) -> [a] -> Stream b -> Stream c

Apply a function to the latest event and the array value at the respective index.

array:                        [ 1, 2, 3 ]
stream:                       --10---10---10---10---10--->
zipItems(add, array, stream): --11---12---13|

The resulting :ref:`Stream` will contain the same number of events as the input :ref:`Stream`, or array.length events, whichever is less.

withItems

withItems :: [a] -> Stream b -> Stream a

Replace each event value with the array item at the respective index.

array:                    [ 1, 2, 3 ]
stream:                   --x--x--x--x--x-->
withItems(array, stream): --1--2--3|

The resulting :ref:`Stream` will contain the same number of events as the input :ref:`Stream`, or array.length events, whichever is less.

Flattening

switchLatest

switchLatest :: Stream (Stream a) -> Stream a

Given a higher-order :ref:`Stream`, return a new :ref:`Stream` that adopts the behavior of (i.e., emits the events of) the most recent inner :ref:`Stream`.

s:                    -a-b-c-d-e-f->
t:                    -1-2-3-4-5-6->
stream:               -s-----t----->
switchLatest(stream): -a-b-c-4-5-6->

join

join :: Stream (Stream a) -> Stream a

Given a higher-order :ref:`Stream`, return a new :ref:`Stream` that merges all the inner :ref:`Streams <Stream>` as they arrive.

s:             ---a---b---c---d-->
t:             -1--2--3--4--5--6->
stream:        -s------t--------->
join(stream):  ---a---b--4c-5-d6->

chain

chain :: (a -> Stream b) -> Stream a -> Stream b

Transform each event in stream into a new :ref:`Stream`, and then merge each into the resulting :ref:`Stream`. Note that f must return a :ref:`Stream`.

stream:            -a----b----c|
f(a):               1--2--3|
f(b):                    1----2----3|
f(c):                           1-2-3|
chain(f, stream):  -1--2-13---2-1-233|

concatMap

concatMap :: (a -> Stream b) -> Stream a -> Stream b

Transform each event in stream into a :ref:`Stream`, and then concatenate each onto the end of the resulting :ref:`Stream`. Note that f must return a :ref:`Stream`.

The mapping function f is applied lazily. That is, f is called only once it is time to concatenate a new stream.

stream:                -a----b----c|
f(a):                   1--2--3|
f(b):                        1----2----3|
f(c):                               1-2-3|
concatMap(f, stream):  -1--2--31----2----31-2-3|
f called lazily:        ^      ^          ^

Note the difference between concatMap and ref:chain: concatMap concatenates, while ref:chain merges.

mergeConcurrently

mergeConcurrently :: int -> Stream (Stream a) -> Stream a

Given a higher-order :ref:`Stream`, return a new :ref:`Stream` that merges inner :ref:`Streams <Stream>` as they arrive up to the specified concurrency. Once concurrency number of :ref:`Streams <Stream>` are being merged, newly arriving :ref:`Streams <Stream>` will be merged after an existing one ends.

s:                            --a--b--c--d--e-->
t:                            --x------y|
u:                            -1--2--3--4--5--6>
stream:                       -s--t--u--------->
mergeConcurrently(2, stream): --a--b--cy4d-5e-6>

Note that u is only merged after t ends because of the concurrency level of 2.

Note also that mergeConcurrently(Infinity, stream) is equivalent to join(stream).

To control concurrency, mergeConcurrently must maintain an internal queue of newly arrived :ref:`Streams <Stream>`. If new :ref:`Streams <Stream>` arrive faster than the concurrency level allows them to be merged, the internal queue will grow infinitely.

mergeMapConcurrently

mergeMapConcurrently :: (a -> Stream b) -> int -> Stream a -> Stream b

Lazily apply a function f to each event in a :ref:`Stream`, merging them into the resulting :ref:`Stream` at the specified concurrency. Once concurrency number of :ref:`Streams <Stream>` are being merged, newly arriving :ref:`Streams <Stream>` will be merged after an existing one ends.

stream:                             --ab--c----d----->
f(a):                               -1-2-3|
f(b):                               -4-5-6----------->
f(c):                               -7--------------->
f(d):                               -1-2-3-4-5-6-7-8->
mergeMapConcurently(f, 2, stream) : ---142536-7------>

Note that f(c) is only merged after f(a) ends.

Also note that f will not get called with d until either f(b) or f(c) ends.

To control concurrency, mergeMapConcurrently must maintain an internal queue of newly arrived :ref:`Streams <Stream>`. If new :ref:`Streams <Stream>` arrive faster than the concurrency level allows them to be merged, the internal queue will grow infinitely.

Merging

merge

merge :: Stream a -> Stream b -> Stream (a | b)

Create a new :ref:`Stream` containing events from two :ref:`Streams <Stream>`.

s1:            -a--b----c--->
s2:            --w---x-y--z->
merge(s1, s2): -aw-b-x-yc-z->

Merging creates a new :ref:`Stream` containing all events from the two original :ref:`Streams <Stream>` without affecting the time of the events. You can think of the events from the input :ref:`Streams <Stream>` simply being interleaved into the new, merged :ref:`Stream`. A merged :ref:`Stream` ends when all of its input :ref:`Streams <Stream>` have ended.

mergeArray

mergeArray :: [ (Stream a) ] -> Stream a

Array form of :ref:`merge`. Create a new :ref:`Stream` containing all events from all :ref:`Streams <Stream>` in the array.

s1:                       -a--b----c---->
s2:                       --w---x-y--z-->
s3:                       ---1---2----3->
mergeArray([s1, s2, s3]): -aw1b-x2yc-z3->

combine

combine :: (a -> b -> c) -> Stream a -> Stream b -> Stream c

Apply a function to the most recent event from each :ref:`Stream` when a new event arrives on any :ref:`Stream`.

s1:                   -0--1----2--->
s2:                   --3---4-5--6->
combine(add, s1, s2): --3-4-5-67-8->

Note that combine waits for at least one event to arrive on all input :ref:`Streams <Stream>` before it produces any events.

combineArray

combineArray :: ((a, b, ...) -> z) -> [ Stream a, Stream b, ... ] -> Stream z

Array form of :ref:`combine`. Apply a function to the most recent event from all :ref:`Streams <Stream>` when a new event arrives on any :ref:`Stream`.

s1:                               -0--1----2->
s2:                               --3---4-5-->
s3:                               ---2---1--->
combineArray(add3, [s1, s2, s3]): ---56-7678->

zip

zip :: (a -> b -> c) -> Stream a -> Stream b -> Stream c

Apply a function to corresponding pairs of events from the inputs :ref:`Streams <Stream>`.

s1:               -1--2--3--4->
s2:               -1---2---3---4->
zip(add, s1, s2): -2---4---6---8->

Zipping correlates by index-corresponding events from two input streams. Note that zipping a "fast" :ref:`Stream` and a "slow" :ref:`Stream` will cause buffering. Events from the fast :ref:`Stream` must be buffered in memory until an event at the corresponding index arrives on the slow :ref:`Stream`.

A zipped :ref:`Stream` ends when any one of its input :ref:`Streams <Stream>` ends.

zipArray

zipArray :: ((a, b, ...) -> z) -> [ Stream a, Stream b, ... ] -> Stream z

Array form of :ref:`zip`. Apply a function to corresponding events from all the inputs :ref:`Streams <Stream>`.

s1:                           -1-2-3---->
s2:                           -1--2--3-->
s3:                           --1--2--3->
zipArray(add3, [s1, s2, s3]): --3--6--9->

_sample

sample

sample :: Stream a -> Stream b -> Stream a

For each event in a sampler :ref:`Stream`, replace the event value with the latest value in another :ref:`Stream`. The resulting :ref:`Stream` will contain the same number of events as the sampler :ref:`Stream`.

values:                  -1--2--3--4--5->
sampler:                 -1-----2-----3->
sample(values, sampler): -1-----3-----5->

values:                  -1-----2-----3->
sampler:                 -1--2--3--4--5->
sample(values, sampler): -1--1--2--2--3->

snapshot

snapshot :: ((a, b) -> c) -> Stream a -> Stream b -> Stream c

For each event in a sampler :ref:`Stream`, apply a function to combine its value with the most recent event value in another :ref:`Stream`. The resulting :ref:`Stream` will contain the same number of events as the sampler :ref:`Stream`.

values:                         -1--2--3--4--5->
sampler:                        -1-----2-----3->
snapshot(sum, values, sampler): -2-----5-----8->

values:                         -1-----2-----3->
sampler:                        -1--2--3--4--5->
snapshot(sum, values, sampler): -2--3--5--6--8->

In contrast to :ref:`combine`, snapshot produces a value only when an event arrives on the sampler.

Filtering

filter

filter :: (a -> bool) -> Stream a -> Stream a

Retain only events for which a predicate is truthy.

stream:               -1-2-3-4->
filter(even, stream): ---2---4->

skipRepeats

skipRepeats :: Stream a -> Stream a

Remove adjacent repeated events.

stream:              -1-2-2-3-4-4-5->
skipRepeats(stream): -1-2---3-4---5->

Note that === is used to identify repeated items. To use a different comparison, use :ref:`skipRepeatsWith`.

skipRepeatsWith

skipRepeatsWith :: ((a, a) -> bool) -> Stream a -> Stream a

Remove adjacent repeated events, using the provided equality function to compare adjacent events.

stream:                                    -a-b-B-c-D-d-e->
skipRepeatsWith(equalsIgnoreCase, stream): -a-b---c-D---e->

The equals function should return true if the two values are equal, or false if they are not equal.

Slicing

slice

slice :: int -> int -> Stream a -> Stream a

Keep only events in a range, where start <= index < end, and index is the ordinal index of an event in stream.

stream:              -a-b-c-d-e-f->
slice(1, 4, stream): ---b-c-d|

stream:              -a-b-c|
slice(1, 4, stream): ---b-c|

If stream contains fewer than start events, the returned :ref:`Stream` will be empty.

take

take :: int -> Stream a -> Stream a

Keep at most the first n events from stream.

stream:          -a-b-c-d-e-f->
take(3, stream): -a-b-c|

stream:          -a-b|
take(3, stream): -a-b|

If stream contains fewer than n events, the returned :ref:`Stream` will effectively be equivalent to stream.

skip

skip :: int -> Stream a -> Stream a

Discard the first n events from stream.

stream:          -a-b-c-d-e-f->
skip(3, stream): -------d-e-f->

stream:          -a-b-c-d-e|
skip(3, stream): -------d-e|

stream:          -a-b-c|
skip(3, stream): ------|

If stream contains fewer than n events, the returned :ref:`Stream` will be empty.

takeWhile

takeWhile :: (a -> bool) -> Stream a -> Stream a

Keep all events until predicate returns false, and discard the rest.

stream:                  -2-4-5-6-8->
takeWhile(even, stream): -2-4-|

skipWhile

skipWhile :: (a -> bool) -> Stream a -> Stream a

Discard all events until predicate returns false, and keep the rest.

stream:                  -2-4-5-6-8->
skipWhile(even, stream): -----5-6-8->

skipAfter

skipAfter :: (a -> bool) -> Stream a -> Stream a

Discard all events after the first event for which predicate returns true.

stream:                  -1-2-3-4-5-6-8->
skipAfter(even, stream): -1-2|

until

until :: Stream * -> Stream a -> Stream a

Keep all events in one :ref:`Stream` until the first event occurs in another.

stream:                   -a-b-c-d-e-f->
endSignal:                ------z->
until(endSignal, stream): -a-b-c|

Note that if endSignal has no events, then the returned :ref:`Stream` will effectively be equivalent to the original.

// Keep only 3 seconds of events, discard the rest.
until(at(3000, null), stream)

since

since :: Stream * -> Stream a -> Stream a

Discard all events in one :ref:`Stream` until the first event occurs in another.

stream:                     -a-b-c-d-e-f->
startSignal:                ------z->
since(startSignal, stream): -------d-e-f->

Note that if startSignal has no events, then the returned :ref:`Stream` will effectively be equivalent to :ref:`never`.

// Discard events for 3 seconds, keep the rest.
since(at(3000, null), stream)

during

during :: Stream (Stream *) -> Stream a -> Stream a

Keep events that occur during a time window defined by a higher-order :ref:`Stream`.

stream:                     -a-b-c-d-e-f-g->
timeWindow:                 -----s
s:                                -----x
during(timeWindow, stream): -----c-d-e-|

This is similar to :ref:`slice`, but uses time rather than indices to "slice" the :ref:`Stream`.

// A time window that:
// 1. starts at time = 1 second
// 2. ends at time = 6 seconds (1 second + 5 seconds).
const timeWindow = at(1000, at(5000, null))

// 1. Discard events for 1 second, then
// 2. keep events for 5 more seconds, then
// 3. discard all subsequent events.
during(timeWindow, stream)

Dealing with time

delay

delay :: Delay -> Stream a -> Stream a

Timeshift a :ref:`Stream` by the specified :ref:`Delay <Delay-type>`.

stream:           -a-b-c-d->
delay(1, stream): --a-b-c-d->
delay(5, stream): ------a-b-c-d->

Delaying a :ref:`Stream` timeshifts all the events by the same amount. It doesn't change the time between events.

withLocalTime

withLocalTime :: Time -> Stream a -> Stream a

Create a Stream with localized :ref:`Time` values, whose origin (i.e., time 0) is at the specified Time on the :ref:`Scheduler` provided when the Stream is observed with :ref:`runEffects` or :ref:`run`.

When implementing custom higher-order :ref:`Stream` combinators, such as :ref:`chain`, you should use withLocalTime to localize "inner" Streams before running them.

Rate limiting

throttle

throttle :: int -> Stream a -> Stream a

Limit the rate of events to at most one per n milliseconds.

stream:               abcd----abcd---->
throttle(2, stream):  a-c-----a-c----->

In contrast to :ref:`debounce`, throttle simply drops events that occur "too often", whereas :ref:`debounce` waits for a "quiet period".

debounce

debounce :: int -> Stream a -> Stream a

Wait for a burst of events to subside and keep only the last event in the burst.

stream:              abcd----abcd---->
debounce(2, stream): -----d-------d-->

If the :ref:`Stream` ends while there is a pending debounced event (e.g., via :ref:`until`), the pending event will occur just before the :ref:`Stream` ends. For example:

s1:                         abcd----abcd---->
s2:                         ------------|
debounce(2, until(s2, s1)): -----d------d|

Debouncing can be extremely useful when dealing with bursts of similar events. For example, debouncing keypress events before initiating a remote search query in a browser application.

const searchInput = document.querySelector('[name="search-text"]');
const searchText = most.fromEvent('input', searchInput);

// The current value of the searchInput, but only
// after the user stops typing for 500 milliseconds.
map(e => e.target.value, debounce(500, searchText))

Dealing with Promises

fromPromise

fromPromise :: Promise a -> Stream a

Create a :ref:`Stream` containing a promise's value.

promise:              ----a
fromPromise(promise): ----a|

If the promise rejects, the :ref:`Stream` will be in an error state with the promise's rejection reason as its error. See :ref:`recoverWith` for error recovery.

awaitPromises

awaitPromises :: Stream (Promise a) -> Stream a

Turn a :ref:`Stream` of promises into a :ref:`Stream` containing the promises' values.

promise p:             ---1
promise q:             ------2
promise r:             -3
stream:                -p---q---r->
awaitPromises(stream): ---1--2--3->

Note that event order is always preserved, regardless of promise fulfillment order.

Using fulfillment order

To create a :ref:`Stream` that merges promises in fulfillment order, use chain(fromPromise, stream). Note the difference:

promise p:                    --1
promise q:                    --------2
promise r:                    ------3
stream:                       -p-q-r----->
chain(fromPromise, stream):   --1---3-2-->
awaitPromises(stream):        --1-----23->

Rejected promises

If a promise rejects, the :ref:`Stream` will be in an error state with the rejected promise's reason as its error. See :ref:`recoverWith` for error recovery. For example:

promise p:             ---1
promise q:             ------X
promise r:             -3
stream:                -p---q---r->
awaitPromises(stream): ---1--X

Forever pending promises

If a promise remains pending forever, the :ref:`Stream` will never produce any events beyond that promise. Use a promise timeout or race in such cases to ensure that all promises either fulfill or reject. For example:

promise p:             ---1
promise q:             ----------->
promise r:             -3
stream:                -p---q---r->
awaitPromises(stream): ---1------->

Handling Errors

recoverWith

recoverWith :: (Error -> Stream a) -> Stream a -> Stream a

Recover from a stream failure by calling a function to create a new :ref:`Stream`.

s:                 -a-b-c-X
f(X):                     d-e-f->
recoverWith(f, s): -a-b-c-d-e-f->

When s fails with an error, f will be called with the error. f must return a new :ref:`Stream` to replace the error.

Sharing Streams

multicast

multicast :: Stream a -> Stream a

Returns a :ref:`Stream` equivalent to the original but which can be shared more efficiently among multiple consumers.

stream:             -a-b-c-d->
multicast(stream):  -a-b-c-d->

Multicast allows you to build up a stream of maps, filters, and other transformations, and then share it efficiently with multiple observers.

Tasks

Helper functions for creating :ref:`Tasks <Task>` to propagate events.

propagateTask

propagateTask :: (Time -> a -> Sink a -> *) -> a -> Sink a -> Task

Create a :ref:`Task` to propagate a value to a :ref:`Sink`. When the :ref:`Task` executes, the provided function will receive the current time (from the :ref:`Scheduler` with which it was scheduled) and the provided value and :ref:`Sink`. The :ref:`Task` can use the :ref:`Sink` to propagate the value in whatever way it chooses. For example as an event or an error, or it could choose not to propagate the event based on some condition, etc.

propagateEventTask

propagateEventTask :: a -> Sink a -> Task

Create a :ref:`Task` that can be scheduled to propagate an event value to a :ref:`Sink`. When the task executes, it will call the :ref:`Sink`'s event method with the current time (from the :ref:`Scheduler` with which it was scheduled) and the value.

propagateEndTask

propagateEndTask :: Sink * -> Task

Create a :ref:`Task` that can be scheduled to propagate end to a :ref:`Sink`. When the task executes, it will call the :ref:`Sink`'s end method with the current time (from the :ref:`Scheduler` with which it was scheduled).

propagateErrorTask

propagateErrorTask :: Error -> Sink * -> Task

Create a :ref:`Task` that can be scheduled to propagate an error to a :ref:`Sink`. When the :ref:`Task` executes, it will call the :ref:`Sink`'s error method with the current time (from the :ref:`Scheduler` with which it was scheduled) and the error.

@most/scheduler

Reading Current Time

currentTime

currentTime :: Scheduler -> Time

Read the current :ref:`Time` from a :ref:`Scheduler`.

Scheduling Tasks

asap

asap :: Task -> Scheduler -> ScheduledTask

Schedule a :ref:`Task` to execute as soon as possible, but still asynchronously.

delay

delay :: Delay -> Task -> Scheduler -> ScheduledTask

Schedule a :ref:`Task` to execute after a specified :ref:`Delay <Delay-type>`.

periodic

periodic :: Period -> Task -> Scheduler -> ScheduledTask

Schedule a :ref:`Task` to execute periodically with the specified :ref:`Period`.

Canceling Tasks

cancelTask

cancelTask :: ScheduledTask -> void

Cancel all future scheduled executions of a :ref:`ScheduledTask`.

cancelAllTasks

Warning

Deprecated: Will be removed in 2.0.0. Instead of using cancelAllTasks, Scheduler callers should track the tasks they create (e.g. by storing them in an array or other data structure), and then cancel each explicitly using :ref:`cancelTask`.

cancelAllTasks :: (ScheduledTask -> boolean) -> Scheduler -> void

Cancel all future scheduled executions of all :ref:`ScheduledTasks <ScheduledTask>` for which the provided predicate is true.

Creating a Scheduler

newScheduler

newScheduler :: Timer -> Timeline -> Scheduler

Create a new :ref:`Scheduler` that uses the provided :ref:`Timer` and :ref:`Timeline` for scheduling :ref:`Tasks <Task>`.

newDefaultScheduler

newDefaultScheduler :: () -> Scheduler

Create a new :ref:`Scheduler` that uses a default platform-specific :ref:`Timer` and a new, empty :ref:`Timeline`.

schedulerRelativeTo

schedulerRelativeTo :: Offset -> Scheduler -> Scheduler

Create a new :ref:`Scheduler` with origin (i.e., zero time) at the specified :ref:`Offset` with the provided :ref:`Scheduler`.

When implementing higher-order :ref:`Stream` combinators, this function can be used to create a :ref:`Scheduler` with local time for each "inner" :ref:`Stream`.

currentTime(scheduler) //> 1637
const relativeScheduler = schedulerRelativeTo(1234, scheduler)
currentTime(relativeScheduler) //> 0

// ... later ...

currentTime(scheduler) //> 3929
currentTime(relativeScheduler) //> 2292

Timer, Timeline, and Clock

newClockTimer

newClockTimer :: Clock -> Timer

Create a new :ref:`Timer` that uses the provided :ref:`Clock` as a source of the current :ref:`Time`.

newTimeline

newTimeline :: () -> Timeline

Create an empty :ref:`Timeline`.

newPlatformClock

newPlatformClock :: () -> Clock

Create a new :ref:`Clock` by auto detecting the best platform-specific source of :ref:`Time`. In modern browsers, it uses performance.now, and on Node, process.hrtime. If neither is available, it falls back to Date.now.

newPerformanceClock

newPerformanceClock :: () -> Clock

Create a new :ref:`Clock` using performance.now.

newHRTimeClock

newHRTimeClock :: () -> Clock

Create a new :ref:`Clock` using process.hrtime.

newDateClock

Warning

Deprecated: Will be removed in 2.0.0. Date.now is not monotonic, and has only been supported as a fallback for browsers that don't support performance.now.

newDateClock :: () -> Clock

Create a new :ref:`Clock` using Date.now. Note that a :ref:`Clock` using Date.now is not guaranteed to be monotonic and is subject to system clock changes, e.g., NTP can change your system clock.

clockRelativeTo

clockRelativeTo :: Clock -> Clock

Create a new :ref:`Clock` whose origin is at the current time (at the instant of calling clockRelativeTime) of the provided :ref:`Clock`.

@most/disposable

Creating Disposables

disposeNone

disposeNone :: () -> Disposable

Create a no-op :ref:`Disposable`.

disposeWith

disposeWith :: (a -> void) -> a -> Disposable

Create a :ref:`Disposable` which, when disposed of, will call the provided function, passing the provided value.

disposeOnce

disposeOnce :: Disposable -> Disposable

Wrap a :ref:`Disposable` so the underlying :ref:`Disposable` will only be disposed of once—even if the returned :ref:`Disposable` is disposed of multiple times.

disposeBoth

disposeBoth :: Disposable -> Disposable -> Disposable

Combine two :ref:`Disposables <Disposable>` into a single :ref:`Disposable` which will dispose of both.

disposeAll

disposeAll :: [Disposable] -> Disposable

Combine an array of :ref:`Disposables <Disposable>` into a single :ref:`Disposable` which will dispose of all the :ref:`Disposables <Disposable>` in the array.

Disposing Disposables

dispose

dispose :: Disposable -> void

Dispose of the provided :ref:`Disposable`. Note that dispose does not catch exceptions. If the :ref:`Disposable` throws an exception, the exception will propagate out of dispose.

tryDispose

tryDispose :: Time -> Disposable -> Sink * -> void

Attempt to dispose of the provided :ref:`Disposable`. If the :ref:`Disposable` throws an exception, catch and propagate it to the provided :ref:`Sink` with the provided :ref:`Time`.

Note: Only an exception thrown by the :ref:`Disposable` will be caught. If the act of propagating an error to the :ref:`Sink` throws an exception, that exception will not be caught.