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 = numberTime 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.
type Delay = numberA Delay represents a duration from "now". When using a default :ref:`Scheduler <newDefaultScheduler>`, the units will be milliseconds.
type Period = numberA Period represents a regular interval. When using a default :ref:`Scheduler <newDefaultScheduler>`, the units will be milliseconds.
type Offset = numberAn 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.
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>`.
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.
type Disposable = {
dispose:: () -> void
}A Disposable represents a resource that must be disposed of (or released), such as a DOM event listener.
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.
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.
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.
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.
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`.
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.
runEffects :: Stream a -> Scheduler -> Promise voidActivate an event :ref:`Stream` and consume all its events.
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 -> DisposableRun 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.
empty :: () -> Stream *Create a :ref:`Stream` containing no events and ends immediately.
empty(): |
never :: () -> Stream *Create a :ref:`Stream` containing no events and never ends.
never(): ---->
now :: a -> Stream aCreate a :ref:`Stream` containing a single event at time 0.
now(x): x|
at :: (Time, a) -> Stream a
at :: Time -> Stream voidCreate a :ref:`Stream` containing a single event at a specific time.
at(3, x): --x|
periodic :: Period -> Stream voidCreate 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 :: Error -> Stream voidCreate 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
newStream :: ((Sink a, Scheduler) -> Disposable) -> Stream aCreate a :ref:`Stream` from a custom event producer (i.e. :ref:`run` method).
import { currentTime as ct } from '@most/scheduler'
// Number -> a -> Stream a
const valueAt = delay => value => newStream(
(sink, scheduler) => {
const publishVal = val => {
sink.event(ct(scheduler), val);
sink.end(ct(scheduler));
};
const timer = setTimeout(publishVal, delay, value);
return { dispose: () => { clearTimeout(timer); } };
}
);
valueAt(1000)(7): ---7|startWith :: a -> Stream a -> Stream aPrepend 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 :: (() -> Stream b) -> Stream a -> Stream (a | b)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`.
map :: (a -> b) -> Stream a -> Stream bApply 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 :: a -> Stream * -> Stream aReplace each event value with x.
stream: -a-b-c-d-> constant(x, stream): -x-x-x-x->
constant('tick', periodic(1000))tap :: (a -> *) -> Stream a -> Stream aPerform 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 :: Stream (a -> b) -> Stream a -> Stream bApply 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 :: (b -> a -> b) -> b -> Stream a -> Stream bIncrementally accumulate results, starting with the provided initial value.
stream: -1-2-3-> scan((x, y) => x + y, 0, stream): 01-3-6->
loop :: (b -> a -> { seed :: b, value :: c }) -> b -> Stream a -> Stream cAccumulate 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 :: ((a, b) -> c) -> [a] -> Stream b -> Stream cApply 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 :: [a] -> Stream b -> Stream aReplace 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.
switchLatest :: Stream (Stream a) -> Stream aGiven 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 :: Stream (Stream a) -> Stream aGiven 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 :: (a -> Stream b) -> Stream a -> Stream bTransform 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 :: (a -> Stream b) -> Stream a -> Stream bTransform 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 :: int -> Stream (Stream a) -> Stream aGiven 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 :: (a -> Stream b) -> int -> Stream a -> Stream bLazily 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.
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 :: [ Stream a, Stream b, ... ] -> Stream (a | b | ...)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 :: (a -> b -> c) -> Stream a -> Stream b -> Stream cApply 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 :: ((a, b, ...) -> z) -> [ Stream a, Stream b, ... ] -> Stream zArray 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 :: (a -> b -> c) -> Stream a -> Stream b -> Stream cApply 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 :: ((a, b, ...) -> z) -> [ Stream a, Stream b, ... ] -> Stream zArray 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 :: Stream a -> Stream b -> Stream aFor 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 :: ((a, b) -> c) -> Stream a -> Stream b -> Stream cFor 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.
filter :: (a -> bool) -> Stream a -> Stream aRetain only events for which a predicate is truthy.
stream: -1-2-3-4-> filter(even, stream): ---2---4->
skipRepeats :: Stream a -> Stream aRemove 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 :: ((a, a) -> bool) -> Stream a -> Stream aRemove 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.
slice :: int -> int -> Stream a -> Stream aKeep 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 :: int -> Stream a -> Stream aKeep 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 :: int -> Stream a -> Stream aDiscard 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 :: (a -> bool) -> Stream a -> Stream aKeep all events until predicate returns false, and discard the rest.
stream: -2-4-5-6-8-> takeWhile(even, stream): -2-4-|
skipWhile :: (a -> bool) -> Stream a -> Stream aDiscard all events until predicate returns false, and keep the rest.
stream: -2-4-5-6-8-> skipWhile(even, stream): -----5-6-8->
skipAfter :: (a -> bool) -> Stream a -> Stream aDiscard 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 :: Stream * -> Stream a -> Stream aKeep 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 :: Stream * -> Stream a -> Stream aDiscard 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 :: Stream (Stream *) -> Stream a -> Stream aKeep 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)delay :: Delay -> Stream a -> Stream aTimeshift 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 :: Time -> Stream a -> Stream aCreate 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.
throttle :: int -> Stream a -> Stream aLimit 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 :: int -> Stream a -> Stream aWait 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))fromPromise :: Promise a -> Stream aCreate 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 :: Stream (Promise a) -> Stream aTurn 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------->
recoverWith :: (Error -> Stream b) -> Stream a -> Stream (a | b)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.
multicast :: Stream a -> Stream aReturns 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.
Helper functions for creating :ref:`Tasks <Task>` to propagate events.
propagateTask :: (Time -> a -> Sink a -> *) -> a -> Sink a -> TaskCreate 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 :: a -> Sink a -> TaskCreate 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 :: Sink * -> TaskCreate 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 :: Error -> Sink * -> TaskCreate 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.
currentTime :: Scheduler -> TimeRead the current :ref:`Time` from a :ref:`Scheduler`.
asap :: Task -> Scheduler -> ScheduledTaskSchedule a :ref:`Task` to execute as soon as possible, but still asynchronously.
delay :: Delay -> Task -> Scheduler -> ScheduledTaskSchedule a :ref:`Task` to execute after a specified :ref:`Delay <Delay-type>`.
periodic :: Period -> Task -> Scheduler -> ScheduledTaskSchedule a :ref:`Task` to execute periodically with the specified :ref:`Period`.
cancelTask :: ScheduledTask -> voidCancel all future scheduled executions of a :ref:`ScheduledTask`.
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 -> voidCancel all future scheduled executions of all :ref:`ScheduledTasks <ScheduledTask>` for which the provided predicate is true.
newScheduler :: Timer -> Timeline -> SchedulerCreate a new :ref:`Scheduler` that uses the provided :ref:`Timer` and :ref:`Timeline` for scheduling :ref:`Tasks <Task>`.
newDefaultScheduler :: () -> SchedulerCreate a new :ref:`Scheduler` that uses a default platform-specific :ref:`Timer` and a new, empty :ref:`Timeline`.
schedulerRelativeTo :: Offset -> Scheduler -> SchedulerCreate 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) //> 2292newClockTimer :: Clock -> TimerCreate a new :ref:`Timer` that uses the provided :ref:`Clock` as a source of the current :ref:`Time`.
newTimeline :: () -> TimelineCreate an empty :ref:`Timeline`.
newPlatformClock :: () -> ClockCreate 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 :: () -> ClockCreate a new :ref:`Clock` using performance.now.
newHRTimeClock :: () -> ClockCreate a new :ref:`Clock` using process.hrtime.
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 :: () -> ClockCreate 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 :: Clock -> ClockCreate a new :ref:`Clock` whose origin is at the current time (at the instant of calling clockRelativeTime) of the provided :ref:`Clock`.
disposeNone :: () -> DisposableCreate a no-op :ref:`Disposable`.
disposeWith :: (a -> void) -> a -> DisposableCreate a :ref:`Disposable` which, when disposed of, will call the provided function, passing the provided value.
disposeOnce :: Disposable -> DisposableWrap 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 :: Disposable -> Disposable -> DisposableCombine two :ref:`Disposables <Disposable>` into a single :ref:`Disposable` which will dispose of both.
disposeAll :: [Disposable] -> DisposableCombine an array of :ref:`Disposables <Disposable>` into a single :ref:`Disposable` which will dispose of all the :ref:`Disposables <Disposable>` in the array.
dispose :: Disposable -> voidDispose 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 :: Time -> Disposable -> Sink * -> voidAttempt 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.