A defalt executor for channel callback dispatching

A convenience alias for take!.

A convenience alias for put!.

Completes at most one of several operations on channel. Receives a vector of operations and optional keyword options.

A channel operation is defined as a vector of 2 elements for take, and 3 elements for put. Unless the :priority option is true and if more than one channel operation is ready, a non-deterministic choice will be made.

Returns a promise instance that will be resolved when a single operation is ready to a vector val channel where val is return value of the operation and channel identifies the channel where the the operation is succeeded.

A blocking variant of alts.

Creates a new channel instance, it optionally accepts buffer, transducer and error handler. If buffer is an integer, it will be used as initial capacity for the expanding buffer.

Returns true if o is instance of Channel or satisfies IChannel protocol.

Close the channel.

A channel exception handler that closes the channel with the cause if an exception is raised in the transducer.

Returns true if channel is closed.

Create a dropping buffer instance.

Create a fixed size (but expanding) buffer instance.

This buffer is used by default if you pass an integer as buffer on channel constructor.

Create a fixed size buffer instance.

Schedules the body to be executed asychronously, potentially using virtual thread if available (a normal thread will be used in other case, determined by executor dynamic var). Returns a promise instance that resolves with the return value when the asynchronous block finishes.

Forwards dynamic bindings.

A convencience go macro version that returns a channel instead of a promise instance, has the same semantics as go macro.

A convencience helper macro that combines go + loop.

Creates an instance of multiplexer.

A multiplexer instance acts like a write-only channel what enables a broadcast-like (instead of a queue-like) behavior. Channels containing copies of this multiplexer can be attached using tap! and dettached with untap!.

Each item is forwarded to all attached channels in parallel and synchronously; use buffers to prevent slow taps from holding up the multiplexer.

If there are no taps, all received items will be dropped. Closed channels will be automatically removed from multiplexer.

Do not creates vthreads (or threads).

Create a multiplexer with an externally provided channel. From now, you can use the external channel or the multiplexer instace to put values in because multiplexer implements the IWriteChannel protocol.

Optionally accepts close? argument, that determines if the channel will be closed when close! is called on multiplexer o not.

Puts a val into channel if it’s possible to do so immediately. Returns a resolved promise with true if the operation succeeded. Never blocks.

Create a once buffer instance.

Puts the contents of coll into the supplied channel.

By default the channel will be closed after the items are copied, but can be determined by the close? parameter. Returns a promise that will be resolved with nil once the items are copied.

Takes elements from the from channel and supplies them to the to channel. By default, the to channel will be closed when the from channel closes, but can be determined by the close? parameter. Will stop consuming the from channel if the to channel closes.

Create a processing pipeline with the ability to specify the process function proc-fn, the type of concurrency primitive to use (:thread or :vthread) and the parallelism.

The proc-fn should be a function that takes the value and the result channel; the result channel should be closed once the processing unit is finished.

By default the output channel is closed when pipeline is terminated, but it can be specified by user using the :close? parameter.

Returns a promise which will be resolved once the pipeline is terminated.

Example:

(def inp (sp/chan)) (def out (sp/chan (map inc)))

(sp/pipeline :typ :vthread :close? true :n 10 :in inp :out out :f proc-fn)

(sp/go (loop [] (when-let val (sp/<! out) (prn “RES:” val) (recur))) (prn “RES: END”))

(p/await! (sp/onto-chan! inp “1” “2” “3” “4” true))

Internally, uses 2 vthreads for pipeline internals processing.

EXPERIMENTAL: API subject to be changed or removed in future releases.

Takes a val from port if it’s possible to do so immediatelly. Returns a resolved promise with the value if succeeded, nil otherwise.

Schedules a put operation on the channel. Returns a promise instance that will resolve to: false if channel is closed, true if put is succeed. If channel has buffer, it will return immediatelly with resolved promise.

Optionally accepts a timeout-duration and timeout-value. The timeout-duration can be a long or Duration instance measured in milliseconds.

A blocking version of put.

Create a sliding buffer instance.

Schedules a take operation on the channel. Returns a promise instance that will resolve to: nil if channel is closed, obj if value is found. If channel has non-empty buffer the take operation will succeed immediatelly with resolved promise.

Optionally accepts a timeout-duration and timeout-value. The timeout-duration can be a long or Duration instance measured in milliseconds.

Blocking version of take.

Copies the multiplexer source onto the provided channel.

A convencience thread macro version that returns a channel instead of a promise instance.

A channel exception handler that throws the exception to the default uncaught exception handler.

Returns a promise that will be resolved in the specified timeout. The default scheduler will be used.

Returns a channel that will be closed in the specified timeout. The default scheduler will be used. You can provide your own as optional first argument.

Disconnects a channel from the multiplexer.