[funcool/beicon "4.1.0"]
beicon is a small and concise library that provides reactive streams API for Clojure and ClojureScript.
This documentation does not cover the whole API, so if you miss some function, contributions are very welcome. You can see the full API documentation here. |
Since beicon is a young project there can be some API breakage.
The simplest way to use beicon in a Clojure project, is by including it in the dependency vector on your project.clj file:
[funcool/beicon "4.1.0"]
This section will give you the available methods for create observable streams.
The most basic way to create a stream is to just take a collection and convert it into an observable sequence:
(require '[beicon.core :as rx])
(def stream (rx/from-coll [1 2 3]))
(rx/on-value stream #(println "v:" %))
;; ==> v: 1
;; ==> v: 2
;; ==> v: 3
An other way to create an observable stream is using the range
constructor,
which is pretty analogous to the Clojures one:
(def stream (rx/range 3))
(rx/on-value stream #(println "v:" %))
;; ==> v: 0
;; ==> v: 1
;; ==> v: 2
Atoms in Clojure are watchable, so you can listen for their changes. This method converts that changes into an infinite observable sequence of atom changes:
(def a (atom 1))
(def stream (rx/from-atom a))
(rx/on-value stream #(println "v:" %))
(swap! a inc)
;; ==> v: 2
It creates an observable sequence of one unique value:
(def stream (rx/just 10)
(rx/on-value stream #(println "v:" %))
;; ==> v: 10
just or once are simple aliases to of function.
|
There is a way to create an observable sequence from
multiple values, using the of
constructor:
(def stream (rx/of 1 2 3))
(rx/on-value stream #(println "v:" %))
;; ==> v: 1
;; ==> v: 2
;; ==> v: 3
Some times you also want just a terminated stream:
(def stream (rx/empty))
This stream does not yield any value and just terminates.
This allows to create an observable sequence of one unique value, that will be emitted after a specified amount of time:
(def stream (rx/timeout 1000 10))
(rx/on-value stream #(println "v:" %))
;; After 1 sec...
;; ==> v: 10
This is the most advanced and flexible way to create an observable sequence. It allows to have control about termination and errors, and is intended to be used for building other kinds of constructors.
(def stream
(rx/create (fn [sink]
(sink 1) ;; next with `1` as value
(sink (rx/end 2)) ;; next with `2` as value and end the stream
(fn []
;; function called on unsubscription
))))
(rx/on-value stream #(println "v:" %))
;; ==> v: 1
;; ==> v: 2
This is implemented using protocols to make it flexible and easily extensible by the user. This is how the default implementation behaves:
Exceptions or exception instances trigger the error termination of stream.
(rx/end value)
sends the unwrapped value to the stream, then terminates stream.
rx/end
as value triggers the stream termination.
nil
triggers stream validation.
any other values are valid values for sending to the stream.
On the JVM, there are other factory methods, that allow creation of backpressure-aware observables:
(def stream
(rx/generate (fn [state sink]
(let [nextval (inc state)]
(sink nextval)
nextval))
(constantly 0)))
The difference to the create
function, the generate
executes the
factory function multiple times and you need to call sink once for each
execution. Additionally, you can maintain state between executions
retuning the desired value and that value will passed as first argument
on the next execution. You can optionally provide an initial state using the
second argument setup
and a third dispose
callback.
Many simple observables can be converted to be backpressure-aware using
the to-flowable
function. Be aware, that the default strategy :buffer
(see below)
is used by default to handle backpressure (over production).
(->> (rx/from-coll [1 2 3])
(rx/to-flowable))
Optionally, you can specify the strategy to use for handling backpressure by the first argument:
(->> (rx/from-coll [1 2 3])
(rx/to-flowable :error))
This is a list of available strategies:
:buffer
- Buffers all values until the downstream consumes it.
:error
- Signals a MissingBackpressureException
in case the downstream can’t keep up.
:drop
- Drops the most recent value, if the downstream can’t keep up.
:latest
- Keeps only the latest value, overwriting any previous value, if the downstream can’t keep up.
The observable sequence can be in three different kind of states: alive, "errored" or ended. If an error is emitted the stream can be considered ended with an error. So error or end states are considered termination states.
And for convenience you can subscribe to any of that states of an observable sequence.
A general purpose subscription is one that allows you to create one subscription, that watches all the different possible states of an observable sequence:
(def sub (rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end:")))
The return value of the subscribe
function is a subscription object, that identifies
the current subscription. It can be cancelled by executing (rx/cancel! sub)
.
But in most circumstances you only want to consume values regardless of any error
or termination. For this purposes there is the on-value
function:
(def sub (rx/on-value sub #(println "val:" %)))
Like with the subscribe
function, the on-value
function also returns a callable, that,
when called, will dispose the created subscription.
take care, that calling any one of those helper functions creates a separated subscription and it can behave unexpectedly, if you are not aware whether you are using hot-vs-cold-observales. |
With the on-end
function you can watch the successful termination of an observable
sequence:
(def sub (rx/on-end sub #(println "end!")))
With the on-error
function you can watch the error termination of an observable
sequence:
(def sub (rx/on-end sub #(println "error:" %)))
The streams created using the generate
function, are backpressure-aware
streams and can be consumed in an backpressure-aware way. For this you need to pass
an object, that implements the ISubscriber
protocol. Lets see an example:
(def stream
(->> (rx/from-coll [1 2 3])
(rx/to-flowable :buffer)))
(rx/subscribe-with stream (reify rx/ISubscriber
(-on-init [_ s]
(s/request! s 1))
(-on-next [_ s v]
(println "on-next:" v)
(s/request! s 1))
(-on-error [_ e]
(println "on-error"))
(-on-end [_]
(println "on-end"))))
;; => on-next: 1
;; => on-next: 2
;; => on-next: 3
;; => on-end
In the same way as the subscribe
function, the subscribe-with
returns
a cancellable subscription object.
The main advantage of using reactive streams is that you may treat them like normal sequences, and in this case filter them with a predicate:
(def stream (->> (rx/from-coll [1 2 3 4 5])
(rx/filter #(> % 3))))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 4
;; ==> on-value: 5
;; ==> on-end
Also, you can apply a function over each value in the stream:
(def stream (->> (rx/from-coll [1 2])
(rx/map inc)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 2
;; ==> on-value: 3
;; ==> on-end
Converts an observable sequence, that can contain other observable sequences, into a new observable sequence, that emits just plain values.
The result is similar to concatenating all the underlying sequences.
(def stream (->> (rx/from-coll [1 2])
(rx/map #(sfrom-coll (range % (+ % 2))))
(rx/flat-map))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 1
;; ==> on-value: 2
;; ==> on-value: 2
;; ==> on-value: 3
;; ==> on-end
Also, sometimes you just want to skip values from stream by different criteria.
You can skip the first N values:
(def stream (->> (rx/from-coll [1 2 3 4 5 6])
(rx/skip 4)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 5
;; ==> on-value: 6
;; ==> on-end
Skip while some predicate evaluates to true
:
(def stream (->> (rx/from-coll [1 1 1 1 2 3])
(rx/skip-while odd?)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 2
;; ==> on-value: 3
;; ==> on-end
Or skip until another observable yields a value with skip-until
(no example at
this moment).
You can also limit the observable sequence to an specified number of elements:
(def stream (->> (rx/from-coll [1 1 1 1 2 3])
(rx/take 2)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 1
;; ==> on-value: 1
;; ==> on-end
Or a predicate evaluates to true
:
(def stream (->> (rx/from-coll [1 1 1 1 2 3])
(rx/take-while odd?)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 1
;; ==> on-value: 1
;; ==> on-value: 1
;; ==> on-value: 1
;; ==> on-end
This is a combination of skip
and take
, and returns an observable
sequence, that represents the portion between start and end of the source
observable sequence.
(def stream (->> (rx/from-coll [1 2 3 4])
(rx/slice 1 3)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 2
;; ==> on-value: 3
;; ==> on-end
Allows combining all results of an observable sequence using a combining function (also called reducing function):
(def stream (->> (rx/from-coll [1 2 3 4])
(rx/reduce + 0)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 10
;; ==> on-end
Like reduce
(see above), but returns a stream of each intermediate result
instead (similar to reductions
in Clojure):
(def stream (->> (rx/from-coll [1 2 3 4])
(rx/scan + 0)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 1
;; ==> on-value: 3
;; ==> on-value: 6
;; ==> on-value: 10
;; ==> on-end
This transformer function allows to accumulate N values in a buffer and then
emits them as one value (similar to partition
in Clojure)
(def stream (->> (rx/from-coll [1 2 3 4])
(rx/buffer 2)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: [1 2]
;; ==> on-value: [3 4]
;; ==> on-end
Performs an arbitrary choice between two or more observable sequences and returns the first value available from any provided observables.
This kind of combinator works very well with operations that can timeout:
(def stream (rx/choice
(rx/timeout 1000 :timeout)
(rx/timeout 900 :value)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: :value
;; ==> on-end
This combinator combines two observable sequences in one.
(def stream (rx/zip
(rx/from-coll [1 2 3])
(rx/from-coll [2 3 4])))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: [1 2]
;; ==> on-value: [2 3]
;; ==> on-value: [3 4]
;; ==> on-end
This combinator concatenates two or more observable sequences in order.
(def stream (rx/concat
(rx/from-coll [1 2])
(rx/from-coll [3 4])))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 1
;; ==> on-value: 2
;; ==> on-value: 3
;; ==> on-value: 4
;; ==> on-end
This combinator merges two or more observable sequences at random (see
concat
for ordered).
(def stream (rx/merge
(rx/from-coll [1 2])
(rx/from-coll [3 4])))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
;; ==> on-value: 1
;; ==> on-value: 3
;; ==> on-value: 2
;; ==> on-value: 4
;; ==> on-end
This is an abstraction that combines observable sequence with the observer. So you can push values into it and transform and subscribe to it like any other sequence.
You can create a subject instance using the subject
constructor function.
This is an example of using subject
for two things: push values and subscribe
to it.
(def subject (rx/subject))
(def stream (->> subject
(rx/skip 1)
(rx/map inc)
(rx/take 2)))
(rx/subscribe stream
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
(rx/push! subject 1)
(rx/push! subject 2)
(rx/push! subject 1)
(rx/push! subject 2)
;; ==> on-value: 3
;; ==> on-value: 2
;; ==> on-end
You can end a subject at any moment just by executing the end!
function:
(def subject (rx/subject))
(rx/subscribe subject
#(println "on-value:" %)
#(println "on-error:" %)
#(println "on-end"))
(rx/end! subject)
;; ==> on-end
Five most important rules:
Beautiful is better than ugly.
Explicit is better than implicit.
Simple is better than complex.
Complex is better than complicated.
Readability counts.
All contributions to beicon should keep these important rules in mind.
Unlike Clojure and other Clojure contributed libraries beicon does not have many restrictions for contributions. Just open an issue or pull request.
beicon is open source and can be found on github.
You can clone the public repository with this command:
git clone https://github.com/funcool/beicon
For running tests just execute this:
./scrpts/build
node ./out/tests.js
lein test
beicon is licensed under BSD (2-Clause) license:
Copyright (c) 2015-2016 Andrey Antukh <niwi@niwi.nz> All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
Can you improve this documentation?Edit on GitHub
cljdoc is a website building & hosting documentation for Clojure/Script libraries
× close