An extremely intuitive, small, and fast functional reactive stream library for JavaScript
xstream is a small, fast, and modular library for reactive programming using streams. It allows you to create, transform, and combine streams of events in a functional and declarative manner.
Creating Streams
This feature allows you to create a stream from a sequence of values. The `xs.of` method creates a stream that emits the values 1, 2, and 3, and then completes.
const xs = require('xstream');
const stream = xs.of(1, 2, 3);
stream.addListener({
next: i => console.log(i),
error: err => console.error(err),
complete: () => console.log('completed')
});Transforming Streams
This feature allows you to transform the values emitted by a stream. The `map` method is used to multiply each value by 2.
const xs = require('xstream');
const stream = xs.of(1, 2, 3).map(x => x * 2);
stream.addListener({
next: i => console.log(i),
error: err => console.error(err),
complete: () => console.log('completed')
});Combining Streams
This feature allows you to combine multiple streams into one. The `combine` method merges the streams and emits an array of the latest values from each stream.
const xs = require('xstream');
const stream1 = xs.of(1, 2, 3);
const stream2 = xs.of('a', 'b', 'c');
const combined = xs.combine(stream1, stream2);
combined.addListener({
next: ([i, j]) => console.log(i, j),
error: err => console.error(err),
complete: () => console.log('completed')
});RxJS is a library for reactive programming using Observables, to make it easier to compose asynchronous or callback-based code. It is more feature-rich and has a larger community compared to xstream, but it is also larger in size.
Most.js is a high-performance reactive programming library that focuses on speed and efficiency. It is similar to xstream in terms of performance but offers a different API and additional features for handling backpressure.
Bacon.js is a functional reactive programming library for JavaScript. It provides a comprehensive set of tools for working with streams and properties, similar to xstream, but with a different API and additional features for handling event streams.
_
__ _____| |_ _ __ ___ __ _ _ __ ___
\ \/ / __| __| '__/ _ \/ _` | '_ ` _ \
> <\__ \ |_| | | __/ (_| | | | | | |
/_/\_\___/\__|_| \___|\__,_|_| |_| |_|
subscribe
import xs from 'xstream'
// Tick every second incremental numbers,
// only pass even numbers, then map them to their square,
// and stop after 5 seconds has passed
var stream = xs.periodic(1000)
.filter(i => i % 2 === 0)
.map(i => i * i)
.endWhen(xs.periodic(5000).take(1))
// So far, the stream is idle.
// As soon as it gets its first listener, it starts executing.
stream.addListener({
next: i => console.log(i),
error: err => console.error(err),
complete: () => console.log('completed'),
})
npm install xstream
import xs from 'xstream'
var xs = require('xstream').default
combineaddListenerremoveListenermapmapTofiltertakedroplaststartWithendWhenfoldreplaceErrorflattenflattenConcurrentlymergecomposerememberimitatedebugshamefullySendNextshamefullySendErrorshamefullySendCompleteXStream has four fundamental types: Stream, Listener, Producer, and MemoryStream.
A Stream is an event emitter with multiple Listeners. When an event happens on the Stream, it is broadcast to all its Listeners at the same time.
Streams have methods attached to them called operators, such as map, filter, fold, take, etc. When called, an operator creates and returns another Stream. The returned Stream is actually a Listener of the source Stream (I forgot to tell you that Streams may be Listeners, too). So once the source Stream broadcasts an event, the event will pass through the operator logic and the returned Stream may perhaps broadcast its own event based on the source one.
You can also trigger an event to happen on a Stream with the shamefullySend* methods. But you don't want to do that. Really, avoid doing that because it's not the reactive way and you'll be missing the point of this library. Ok?
A Listener is an object with three functions attached to it: next, error, and complete. There is one function for each type of event a Stream may emit.
next events are the typical type, they deliver a value.error events abort (stop) the execution of the Stream, and happen when something goes wrong in the Stream (or upstream somewhere in the chain of operators)complete events signal the peaceful stop of the execution of the Stream.This is an example of a typical listener:
var listener = {
next: (value) => {
console.log('The Stream gave me a value: ', value);
},
error: (err) => {
console.error('The Stream gave me an error: ', err);
},
complete: () => {
console.log('The Stream told me it is done.');
},
}
And this is how you would attach that Listener to a Stream:
stream.addListener(listener)
And when you think the Listener is done, you can remove it from the Stream:
stream.removeListener(listener)
A Producer is like a machine that produces events to be broadcast on a Stream.
Events from a Stream must come from somewhere, right? That's why we need Producers. They are objects with two functions attached: start(listener) and stop(). Once you call start with a listener, the Producer will start generating events and it will send those to the listener. When you call stop(), the Producer should quit doing its own thing.
Because Streams are Listeners, if you give a Stream as the Listener in start(stream), essentially the Producer is now generating events that will be broadcast on the Stream. Nice, huh? Now a bunch of listeners can be attached to the Stream and they can all get those events originally coming from the Producer. That's why xs.create(producer) receives a Producer to be the heart of a new Stream. Check this out:
var producer = {
start: function (listener) {
this.id = setInterval(() => listener.next('yo'), 1000)
},
stop: function () {
clearInterval(this.id)
},
id: 0,
}
// This fellow delivers a 'yo' next event every 1 second
var stream = xs.create(producer)
But remember, a Producer has only one listener, but a Stream may have many listeners.
You may wonder "when is start and stop called", and that's actually a fairly tricky topic, so let's get back to that soon. First let me tell you about MemoryStreams.
A MemoryStream is just like a Stream: it has operators, it can have listeners attached, you can shamefully send events to it, blabla. But it has one special property: it has memory. It remembers the most recent (but just one) next event that it sent to its listeners.
Why is that useful? If a new Listener is added after that next event was sent, the MemoryStream will get its value stored in memory and will send it to the newly attached Listener.
This is important so MemoryStreams can represent values or pieces of state which are relevant even after some time has passed. You don't want to lose those, you want to keep them and send them to Listeners that arrive late, after the event was originally created.
A Stream controls its Producer according to its number of Listeners, using reference counting with a synchronous start and a cancelable asynchronous stop. That's how a Stream starts and stops, basically. Usually this part of XStream is not so relevant to remember when building applications, but if you want to understand it for debugging or curiosity, it's explained in plain English below.
When you create a Stream with xs.create(producer), the start() function of the Producer is not yet called. The Stream is still "idle". It has the Producer, but the Producer was not turned on.
Once the first Listener is added to the Stream, the number of Listeners attached suddenly changed from 0 to 1. That's when the Stream calls start, because after all there is at least one Listener interested in this Stream.
More Listeners may be added in the future, but they don't affect whether the Producer will continue working or stop. Just the first Listener dictates when the Stream starts.
What matters for stopping the Producer is stream.removeListener. When the last Listener leaves (or in other words, when the number of Listeners suddenly changes from 1 to 0), the Stream schedules producer.stop() to happen on the next event loop. That is, asynchronously. If, however, a new Listener is added (number goes from 0 to 1) before that scheduled moment, the producer.stop() will be cancelled, and the Producer will continue generating events for its Stream normally.
The reason the Producer is not suddenly (synchronously) stopped, is that it is often necessary to swap the single listener of a Stream, but still keep its ongoing execution. For instance:
var listenerA = {/* ... */}
var listenerB = {/* ... */}
// number goes from 0 to 1, so the Stream's Producer starts
stream.addListener(listenerA)
// ...
// number goes from 1 to 0, but then immediately goes back
// to 1, because listenerB was added
stream.removeListener(listenerA)
stream.addListener(listenerB)
// Stream's Producer does not stop, everything continues as before
It's still useful to eventually (asynchronously) stop a Stream's internal Producer, because you don't want useless computation lying around producing gibberish. At least I don't.
Factories are functions that create Streams, such as xs.create(), xs.periodic(), etc.
create(producer)Creates a new Stream given a Producer.
producer: Producer An optional Producer that dictates how to start, generate events, and stop the Stream.createWithMemory(producer)Creates a new MemoryStream given a Producer.
producer: Producer An optional Producer that dictates how to start, generate events, and stop the Stream.never()Creates a Stream that does nothing when started. It never emits any event.
Marble diagram:
never
-----------------------
empty()Creates a Stream that immediately emits the "complete" notification when started, and that's it.
Marble diagram:
empty
-|
throw(error)Creates a Stream that immediately emits an "error" notification with the
value you passed as the error argument when the stream starts, and that's
it.
Marble diagram:
throw(X)
-X
error The error event to emit on the created stream.of(a, b)Creates a Stream that immediately emits the arguments that you give to of, then completes.
Marble diagram:
of(1,2,3)
123|
a The first value you want to emit as an event on the stream.b The second value you want to emit as an event on the stream. One or more of these values may be given as arguments.fromArray(array)Converts an array to a stream. The returned stream will emit synchronously all the items in the array, and then complete.
Marble diagram:
fromArray([1,2,3])
123|
array: Array The array to be converted as a stream.fromPromise(promise)Converts a promise to a stream. The returned stream will emit the resolved value of the promise, and then complete. However, if the promise is rejected, the stream will emit the corresponding error.
Marble diagram:
fromPromise( ----42 )
-----------------42|
promise: Promise The promise to be converted as a stream.periodic(period)Creates a stream that periodically emits incremental numbers, every
period milliseconds.
Marble diagram:
periodic(1000)
---0---1---2---3---4---...
period: number The interval in milliseconds to use as a rate of emission.merge(stream1, stream2)Blends multiple streams together, emitting events from all of them concurrently.
merge takes multiple streams as arguments, and creates a stream that imitates each of the argument streams, in parallel.
Marble diagram:
--1----2-----3--------4---
----a-----b----c---d------
merge
--1-a--2--b--3-c---d--4---
stream1: Stream A stream to merge together with other streams.stream2: Stream A stream to merge together with other streams. Two or more streams may be given as arguments.combine(project, stream1, stream2)Combines multiple streams together to return a stream whose events are calculated from the latest events of each of the input streams.
combine remembers the most recent event from each of the input streams.
When any of the input streams emits an event, that event together with all
the other saved events are combined in the project function which should
return a value. That value will be emitted on the output stream. It's
essentially a way of mixing the events from multiple streams according to a
formula.
Marble diagram:
--1----2-----3--------4---
----a-----b-----c--d------
combine((x,y) => x+y)
----1a-2a-2b-3b-3c-3d-4d--
project: Function A function of type (x: T1, y: T2) => R or similar that takes the most recent events x and y from the input
streams and returns a value. The output stream will emit that value. The
number of arguments for this function should match the number of input
streams.stream1: Stream A stream to combine together with other streams.stream2: Stream A stream to combine together with other streams. Two or more streams may be given as arguments.Methods are functions attached to a Stream instance, like stream.addListener(). Operators are also methods, but return a new Stream, leaving the existing Stream unmodified, except for the fact that it has a child Stream attached as Listener.
combine(project, other)Combines multiple streams with the input stream to return a stream whose events are calculated from the latest events of each of its input streams.
combine remembers the most recent event from each of the input streams.
When any of the input streams emits an event, that event together with all
the other saved events are combined in the project function which should
return a value. That value will be emitted on the output stream. It's
essentially a way of mixing the events from multiple streams according to a
formula.
Marble diagram:
--1----2-----3--------4---
----a-----b-----c--d------
combine((x,y) => x+y)
----1a-2a-2b-3b-3c-3d-4d--
project: Function A function of type (x: T1, y: T2) => R or similar that takes the most recent events x and y from the input
streams and returns a value. The output stream will emit that value. The
number of arguments for this function should match the number of input
streams.other: Stream Another stream to combine together with the input stream. There may be more of these arguments.addListener(listener)Adds a Listener to the Stream.
listener: Listener\<T>removeListener(listener)Removes a Listener from the Stream, assuming the Listener was added to it.
listener: Listener\<T>map(project)Transforms each event from the input Stream through a project function,
to get a Stream that emits those transformed events.
Marble diagram:
--1---3--5-----7------
map(i => i * 10)
--10--30-50----70-----
project: Function A function of type (t: T) => U that takes event t of type T from the input Stream and produces an event of type U, to
be emitted on the output Stream.mapTo(projectedValue)It's like map, but transforms each input event to always the same
constant value on the output Stream.
Marble diagram:
--1---3--5-----7-----
mapTo(10)
--10--10-10----10----
projectedValue A value to emit on the output Stream whenever the input Stream emits any value.filter(passes)Only allows events that pass the test given by the passes argument.
Each event from the input stream is given to the passes function. If the
function returns true, the event is forwarded to the output stream,
otherwise it is ignored and not forwarded.
Marble diagram:
--1---2--3-----4-----5---6--7-8--
filter(i => i % 2 === 0)
------2--------4---------6----8--
passes: Function A function of type (t: T) +> boolean that takes an event from the input stream and checks if it passes, by returning a
boolean.take(amount)Lets the first amount many events from the input stream pass to the
output stream, then makes the output stream complete.
Marble diagram:
--a---b--c----d---e--
take(3)
--a---b--c|
amount: number How many events to allow from the input stream before completing the output stream.drop(amount)Ignores the first amount many events from the input stream, and then
after that starts forwarding events from the input stream to the output
stream.
Marble diagram:
--a---b--c----d---e--
drop(3)
--------------d---e--
amount: number How many events to ignore from the input stream before forwarding all events from the input stream to the output stream.last()When the input stream completes, the output stream will emit the last event emitted by the input stream, and then will also complete.
Marble diagram:
--a---b--c--d----|
last()
-----------------d|
startWith(initial)Prepends the given initial value to the sequence of events emitted by the
input stream.
Marble diagram:
---1---2-----3---
startWith(0)
0--1---2-----3---
initial The value or event to prepend.endWhen(other)Uses another stream to determine when to complete the current stream.
When the given other stream emits an event or completes, the output
stream will complete. Before that happens, the output stream will imitate
whatever happens on the input stream.
Marble diagram:
---1---2-----3--4----5----6---
endWhen( --------a--b--| )
---1---2-----3--4--|
other Some other stream that is used to know when should the output stream of this operator complete.fold(accumulate, seed)"Folds" the stream onto itself.
Combines events from the past throughout
the entire execution of the input stream, allowing you to accumulate them
together. It's essentially like Array.prototype.reduce.
The output stream starts by emitting the seed which you give as argument.
Then, when an event happens on the input stream, it is combined with that
seed value through the accumulate function, and the output value is
emitted on the output stream. fold remembers that output value as acc
("accumulator"), and then when a new input event t happens, acc will be
combined with that to produce the new acc and so forth.
Marble diagram:
------1-----1--2----1----1------
fold((acc, x) => acc + x, 3)
3-----4-----5--7----8----9------
accumulate: Function A function of type (acc: R, t: T) => R that takes the previous accumulated value acc and the incoming event from the
input stream and produces the new accumulated value.seed The initial accumulated value, of type R.replaceError(replace)Replaces an error with another stream.
When (and if) an error happens on the input stream, instead of forwarding
that error to the output stream, replaceError will call the replace
function which returns the stream that the output stream will imitate. And,
in case that new stream also emits an error, replace will be called again
to get another stream to start imitating.
Marble diagram:
--1---2-----3--4-----X
replaceError( () => --10--| )
--1---2-----3--4--------10--|
replace: Function A function of type (err) => Stream that takes the error that occurred on the input stream or on the previous replacement
stream and returns a new stream. The output stream will imitate the stream
that this function returns.flatten()Flattens a "stream of streams", handling only one nested stream at a time (no concurrency).
If the input stream is a stream that emits streams, then this operator will return an output stream which is a flat stream: emits regular events. The flattening happens without concurrency. It works like this: when the input stream emits a nested stream, flatten will start imitating that nested one. However, as soon as the next nested stream is emitted on the input stream, flatten will forget the previous nested one it was imitating, and will start imitating the new nested one.
Marble diagram:
--+--------+---------------
\ \
\ ----1----2---3--
--a--b----c----d--------
flatten
-----a--b------1----2---3--
flattenConcurrently()Flattens a "stream of streams", handling multiple concurrent nested streams simultaneously.
If the input stream is a stream that emits streams, then this operator will return an output stream which is a flat stream: emits regular events. The flattening happens concurrently. It works like this: when the input stream emits a nested stream, flattenConcurrently will start imitating that nested one. When the next nested stream is emitted on the input stream, flattenConcurrently will also imitate that new one, but will continue to imitate the previous nested streams as well.
Marble diagram:
--+--------+---------------
\ \
\ ----1----2---3--
--a--b----c----d--------
flattenConcurrently
-----a--b----c-1--d-2---3--
merge(other)Blends two streams together, emitting events from both.
merge takes an other stream and returns an output stream that imitates
both the input stream and the other stream.
Marble diagram:
--1----2-----3--------4---
----a-----b----c---d------
merge
--1-a--2--b--3-c---d--4---
other: Stream Another stream to merge together with the input stream.compose(operator)Passes the input stream to a custom operator, to produce an output stream.
compose is a handy way of using an existing function in a chained style.
Instead of writing outStream = f(inStream) you can write
outStream = inStream.compose(f).
operator: function A function that takes a stream as input and returns a stream as well.remember()Returns an output stream that imitates the input stream, but also remembers the most recent event that happens on the input stream, so that a newly added listener will immediately receive that memorised event.
imitate(other)Changes this current stream to imitate the other given stream.
The imitate method returns nothing. Instead, it changes the behavior of
the current stream, making it re-emit whatever events are emitted by the
given other stream.
other: Stream The stream to imitate on the current one.debug(labelOrSpy)Returns an output stream that identically imitates the input stream, but
also runs a spy function fo each event, to help you debug your app.
debug takes a spy function as argument, and runs that for each event
happening on the input stream. If you don't provide the spy argument,
then debug will just console.log each event. This helps you to
understand the flow of events through some operator chain.
Please note that if the output stream has no listeners, then it will not
start, which means spy will never run because no actual event happens in
that case.
Marble diagram:
--1----2-----3-----4--
debug
--1----2-----3-----4--
labelOrSpy: function A string to use as the label when printing debug information on the console, or a 'spy' function that takes an event
as argument, and does not need to return anything.shamefullySendNext(value)Forces the Stream to emit the given value to its listeners.
As the name indicates, if you use this, you are most likely doing something The Wrong Way. Please try to understand the reactive way before using this method. Use it only when you know what you are doing.
value The "next" value you want to broadcast to all listeners of this Stream.shamefullySendError(error)Forces the Stream to emit the given error to its listeners.
As the name indicates, if you use this, you are most likely doing something The Wrong Way. Please try to understand the reactive way before using this method. Use it only when you know what you are doing.
error: any The error you want to broadcast to all the listeners of this Stream.shamefullySendComplete()Forces the Stream to emit the "completed" event to its listeners.
As the name indicates, if you use this, you are most likely doing something The Wrong Way. Please try to understand the reactive way before using this method. Use it only when you know what you are doing.
The following are standalone stream operators and stream factories that may be separately imported and utilized in your project.
To use an extra operator (e.g. delay), import it as such:
import xs from 'xstream'
import delay from 'xstream/extra/delay'
const inputStream = xs.of(1, 2, 3, 4)
const outputStream = inputStream.compose(delay(500))
To import and use an extra factory (e.g. fromEvent), import it as such:
import fromEvent from 'xstream/extra/fromEvent'
const clickStream = fromEvent(document, 'click')
concat(stream1, stream2)Puts one stream after the other. concat is a factory that takes multiple
streams as arguments, and starts the n+1-th stream only when the n-th
stream has completed. It concatenates those streams together.
Marble diagram:
--1--2---3---4-|
...............--a-b-c--d-|
concat
--1--2---3---4---a-b-c--d-|
Example:
import concat from 'xstream/extra/concat'
const streamA = xs.of('a', 'b', 'c')
const streamB = xs.of(10, 20, 30)
const streamC = xs.of('X', 'Y', 'Z')
const outputStream = concat(streamA, streamB, streamC)
outputStream.addListener({
next: (x) => console.log(x),
error: (err) => console.error(err),
complete: () => console.log('concat completed'),
})
stream1: Stream A stream to concatenate together with other streams.stream2: Stream A stream to concatenate together with other streams. Two or more streams may be given as arguments.fromDiagram(diagram, options)Creates a real stream out of an ASCII drawing of a stream. Each string
character represents an amount of time passed (by default, 20 milliseconds).
- characters represent nothing special, | is a symbol to mark the
completion of the stream, # is an error on the stream, and any other
character is a "next" event.
Example:
import fromDiagram from 'xstream/extra/fromDiagram'
const stream = fromDiagram('--a--b---c-d--|')
stream.addListener({
next: (x) => console.log(x),
error: (err) => console.error(err),
complete: () => console.log('concat completed'),
})
The character a represent emission of the event 'a', a string. If you
want to emit something else than a string, you need to provide those values
in the options argument.
Example:
import fromDiagram from 'xstream/extra/fromDiagram'
const stream = fromDiagram('--a--b---c-d--|', {
values: {a: 10, b: 20, c: 30, d: 40}
})
stream.addListener({
next: (x) => console.log(x),
error: (err) => console.error(err),
complete: () => console.log('concat completed'),
})
That way, the stream will emit the numbers 10, 20, 30, 40. The options
argument may also take timeUnit, a number to configure how many
milliseconds does each represents, and errorValue, a value to send out as
the error which # represents.
diagram: string A string representing a timeline of values, error, or complete notifications that should happen on the output stream.options An options object that allows you to configure some additional details of the creation of the stream.Q: What's the difference between xstream and RxJS?
A: Read this blog post on the topic.
Q: What is the equivalent of withLatestFrom in xstream?
A.withLatestFrom(B, (a, b) => a + b)
can be achieved in xstream with
B.map(b => A.map(a => a + b)).flatten()
And can be interpreted/read as "when a B event happens, remember it and map it to all the subsequent events of A mapped to a + b".
Acknowledgements: xstream is built by staltz and TylorS.
License: MIT
FAQs
An extremely intuitive, small, and fast functional reactive stream library for JavaScript
The npm package xstream receives a total of 97,047 weekly downloads. As such, xstream popularity was classified as popular.
We found that xstream demonstrated a not healthy version release cadence and project activity because the last version was released a year ago. It has 2 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Research
An advanced npm supply chain attack is leveraging Ethereum smart contracts for decentralized, persistent malware control, evading traditional defenses.
Security News
Research
Attackers are impersonating Sindre Sorhus on npm with a fake 'chalk-node' package containing a malicious backdoor to compromise developers' projects.
Security News
The npm package for the LottieFiles Player web component was hit with a supply chain attack after a software engineer's npmjs credentials were compromised.