API for clojure.core.async - core.async 1.7.702-SNAPSHOT (in development)


Full namespace name: clojure.core.async

Overview

Facilities for async programming and communication.

go blocks are dispatched over an internal thread pool, which
defaults to 8 threads. The size of this pool can be modified using
the Java system property `clojure.core.async.pool-size`.

Set Java system property `clojure.core.async.go-checking` to true
to validate go blocks do not invoke core.async blocking operations.
Property is read once, at namespace load time. Recommended for use
primarily during development. Invalid blocking calls will throw in
go block threads - use Thread.setDefaultUncaughtExceptionHandler()
to catch and handle.

Public Variables and Functions



<!

function
Usage: (<! port)
takes a val from port. Must be called inside a (go ...) block. Will
return nil if closed. Will park if nothing is available.

    
    
    Source
  


<!!

function
Usage: (<!! port)
takes a val from port. Will return nil if closed. Will block
if nothing is available.
Not intended for use in direct or transitive calls from (go ...) blocks.
Use the clojure.core.async.go-checking flag to detect invalid use (see
namespace docs).

    
    
    Source
  


>!

function
Usage: (>! port val)
puts a val into port. nil values are not allowed. Must be called
inside a (go ...) block. Will park if no buffer space is available.
Returns true unless port is already closed.

    
    
    Source
  


>!!

function
Usage: (>!! port val)
puts a val into port. nil values are not allowed. Will block if no
buffer space is available. Returns true unless port is already closed.
Not intended for use in direct or transitive calls from (go ...) blocks.
Use the clojure.core.async.go-checking flag to detect invalid use (see
namespace docs).

    
    
    Source
  


admix

function
Usage: (admix mix ch)
Adds ch as an input to the mix

    
    
    Source
  


alt!

macro
Usage: (alt! & clauses)
Makes a single choice between one of several channel operations,
as if by alts!, returning the value of the result expr corresponding
to the operation completed. Must be called inside a (go ...) block.

Each clause takes the form of:

channel-op[s] result-expr

where channel-ops is one of:

take-port - a single port to take
[take-port | [put-port put-val] ...] - a vector of ports as per alts!
:default | :priority - an option for alts!

and result-expr is either a list beginning with a vector, whereupon that
vector will be treated as a binding for the [val port] return of the
operation, else any other expression.

(alt!
  [c t] ([val ch] (foo ch val))
  x ([v] v)
  [[out val]] :wrote
  :default 42)

Each option may appear at most once. The choice and parking
characteristics are those of alts!.

    
    
    Source
  


alt!!

macro
Usage: (alt!! & clauses)
Like alt!, except as if by alts!!, will block until completed, and
not intended for use in (go ...) blocks.

    
    
    Source
  


alts!

function
Usage: (alts! ports & {:as opts})
Completes at most one of several channel operations. Must be called
inside a (go ...) block. ports is a vector of channel endpoints,
which can be either a channel to take from or a vector of
[channel-to-put-to val-to-put], in any combination. Takes will be
made as if by <!, and puts will be made as if by >!. Unless
the :priority option is true, if more than one port operation is
ready a non-deterministic choice will be made. If no operation is
ready and a :default value is supplied, [default-val :default] will
be returned, otherwise alts! will park until the first operation to
become ready completes. Returns [val port] of the completed
operation, where val is the value taken for takes, and a
boolean (true unless already closed, as per put!) for puts.

opts are passed as :key val ... Supported options:

:default val - the value to use if none of the operations are immediately ready
:priority true - (default nil) when true, the operations will be tried in order.

Note: there is no guarantee that the port exps or val exprs will be
used, nor in what order should they be, so they should not be
depended upon for side effects.

    
    
    Source
  


alts!!

function
Usage: (alts!! ports & opts)
Like alts!, except takes will be made as if by <!!, and puts will
be made as if by >!!, will block until completed.
Not intended for use in direct or transitive calls from (go ...) blocks.
Use the clojure.core.async.go-checking flag to detect invalid use (see
namespace docs).

    
    
    Source
  


buffer

function
Usage: (buffer n)
Returns a fixed buffer of size n. When full, puts will block/park.

    
    
    Source
  


chan

function
Usage: (chan)
       (chan buf-or-n)
       (chan buf-or-n xform)
       (chan buf-or-n xform ex-handler)
Creates a channel with an optional buffer, an optional transducer
(like (map f), (filter p) etc or a composition thereof), and an
optional exception-handler.  If buf-or-n is a number, will create
and use a fixed buffer of that size. If a transducer is supplied a
buffer must be specified. ex-handler must be a fn of one argument -
if an exception occurs during transformation it will be called with
the Throwable as an argument, and any non-nil return value will be
placed in the channel.

    
    
    Source
  


close!

function
Usage: (close! chan)
Closes a channel. The channel will no longer accept any puts (they
will be ignored). Data in the channel remains available for taking, until
exhausted, after which takes will return nil. If there are any
pending takes, they will be dispatched with nil. Closing a closed
channel is a no-op. Returns nil.

Logically closing happens after all puts have been delivered. Therefore, any
blocked or parked puts will remain blocked/parked until a taker releases them.

    
    
    Source
  


do-alts

function
Usage: (do-alts fret ports opts)
returns derefable [val port] if immediate, nil if enqueued

    
    
    Source
  


dropping-buffer

function
Usage: (dropping-buffer n)
Returns a buffer of size n. When full, puts will complete but
val will be dropped (no transfer).

    
    
    Source
  


go

macro
Usage: (go & body)
Asynchronously executes the body, returning immediately to the
calling thread. Additionally, any visible calls to <!, >! and alt!/alts!
channel operations within the body will block (if necessary) by
'parking' the calling thread rather than tying up an OS thread (or
the only JS thread when in ClojureScript). Upon completion of the
operation, the body will be resumed.

go blocks should not (either directly or indirectly) perform operations
that may block indefinitely. Doing so risks depleting the fixed pool of
go block threads, causing all go block processing to stop. This includes
core.async blocking ops (those ending in !!) and other blocking IO.

Returns a channel which will receive the result of the body when
completed

    
    
    Source
  


go-loop

macro
Usage: (go-loop bindings & body)
Like (go (loop ...))

    
    
    Source
  


into

function
Usage: (into coll ch)
Returns a channel containing the single (collection) result of the
items taken from the channel conjoined to the supplied
collection. ch must close before into produces a result.

    
    
    Source
  


map

function
Usage: (map f chs)
       (map f chs buf-or-n)
Takes a function and a collection of source channels, and returns a
channel which contains the values produced by applying f to the set
of first items taken from each source channel, followed by applying
f to the set of second items from each channel, until any one of the
channels is closed, at which point the output channel will be
closed. The returned channel will be unbuffered by default, or a
buf-or-n can be supplied

    
    
    Source
  


merge

function
Usage: (merge chs)
       (merge chs buf-or-n)
Takes a collection of source channels and returns a channel which
contains all values taken from them. The returned channel will be
unbuffered by default, or a buf-or-n can be supplied. The channel
will close after all the source channels have closed.

    
    
    Source
  


mix

function
Usage: (mix out)
Creates and returns a mix of one or more input channels which will
be put on the supplied out channel. Input sources can be added to
the mix with 'admix', and removed with 'unmix'. A mix supports
soloing, muting and pausing multiple inputs atomically using
'toggle', and can solo using either muting or pausing as determined
by 'solo-mode'.

Each channel can have zero or more boolean modes set via 'toggle':

:solo - when true, only this (ond other soloed) channel(s) will appear
        in the mix output channel. :mute and :pause states of soloed
        channels are ignored. If solo-mode is :mute, non-soloed
        channels are muted, if :pause, non-soloed channels are
        paused.

:mute - muted channels will have their contents consumed but not included in the mix
:pause - paused channels will not have their contents consumed (and thus also not included in the mix)

    
    
    Source
  


mult

function
Usage: (mult ch)
Creates and returns a mult(iple) of the supplied channel. Channels
containing copies of the channel can be created with 'tap', and
detached with 'untap'.

Each item is distributed to all taps in parallel and synchronously,
i.e. each tap must accept before the next item is distributed. Use
buffering/windowing to prevent slow taps from holding up the mult.

Items received when there are no taps get dropped.

If a tap puts to a closed channel, it will be removed from the mult.

    
    
    Source
  


offer!

function
Usage: (offer! port val)
Puts a val into port if it's possible to do so immediately.
nil values are not allowed. Never blocks. Returns true if offer succeeds.

    
    
    Source
  


onto-chan

function
Usage: (onto-chan ch coll)
       (onto-chan ch coll close?)
Deprecated - use onto-chan! or onto-chan!!

    
    Deprecated since core.async version 1.2
Source


onto-chan!

function
Usage: (onto-chan! ch coll)
       (onto-chan! ch coll close?)
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 channel which will close after the items are copied.

If accessing coll might block, use onto-chan!! instead

    
    
    Source
  


onto-chan!!

function
Usage: (onto-chan!! ch coll)
       (onto-chan!! ch coll close?)
Like onto-chan! for use when accessing coll might block,
e.g. a lazy seq of blocking operations

    
    
    Source
  


pipe

function
Usage: (pipe from to)
       (pipe from to close?)
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

    
    
    Source
  


pipeline

function
Usage: (pipeline n to xf from)
       (pipeline n to xf from close?)
       (pipeline n to xf from close? ex-handler)
Takes elements from the from channel and supplies them to the to
channel, subject to the transducer xf, with parallelism n. Because
it is parallel, the transducer will be applied independently to each
element, not across elements, and may produce zero or more outputs
per input.  Outputs will be returned in order relative to the
inputs. 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. Note this
should be used for computational parallelism. If you have multiple
blocking operations to put in flight, use pipeline-blocking instead,
If you have multiple asynchronous operations to put in flight, use
pipeline-async instead. See chan for semantics of ex-handler.

    
    
    Source
  


pipeline-async

function
Usage: (pipeline-async n to af from)
       (pipeline-async n to af from close?)
Takes elements from the from channel and supplies them to the to
channel, subject to the async function af, with parallelism n. af
must be a function of two arguments, the first an input value and
the second a channel on which to place the result(s). The
presumption is that af will return immediately, having launched some
asynchronous operation whose completion/callback will put results on
the channel, then close! it. Outputs will be returned in order
relative to the inputs. 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. See also pipeline, pipeline-blocking.

    
    
    Source
  


pipeline-blocking

function
Usage: (pipeline-blocking n to xf from)
       (pipeline-blocking n to xf from close?)
       (pipeline-blocking n to xf from close? ex-handler)
Like pipeline, for blocking operations.

    
    
    Source
  


poll!

function
Usage: (poll! port)
Takes a val from port if it's possible to do so immediately.
Never blocks. Returns value if successful, nil otherwise.

    
    
    Source
  


promise-chan

function
Usage: (promise-chan)
       (promise-chan xform)
       (promise-chan xform ex-handler)
Creates a promise channel with an optional transducer, and an optional
exception-handler. A promise channel can take exactly one value that consumers
will receive. Once full, puts complete but val is dropped (no transfer).
Consumers will block until either a value is placed in the channel or the
channel is closed, then return the value (or nil) forever. See chan for the
semantics of xform and ex-handler.

    
    
    Source
  


pub

function
Usage: (pub ch topic-fn)
       (pub ch topic-fn buf-fn)
Creates and returns a pub(lication) of the supplied channel,
partitioned into topics by the topic-fn. topic-fn will be applied to
each value on the channel and the result will determine the 'topic'
on which that value will be put. Channels can be subscribed to
receive copies of topics using 'sub', and unsubscribed using
'unsub'. Each topic will be handled by an internal mult on a
dedicated channel. By default these internal channels are
unbuffered, but a buf-fn can be supplied which, given a topic,
creates a buffer with desired properties.

Each item is distributed to all subs in parallel and synchronously,
i.e. each sub must accept before the next item is distributed. Use
buffering/windowing to prevent slow subs from holding up the pub.

Items received when there are no matching subs get dropped.

Note that if buf-fns are used then each topic is handled
asynchronously, i.e. if a channel is subscribed to more than one
topic it should not expect them to be interleaved identically with
the source.

    
    
    Source
  


put!

function
Usage: (put! port val)
       (put! port val fn1)
       (put! port val fn1 on-caller?)
Asynchronously puts a val into port, calling fn1 (if supplied) when
complete, passing false iff port is already closed. nil values are
not allowed. If on-caller? (default true) is true, and the put is
immediately accepted, will call fn1 on calling thread.

fn1 may be run in a fixed-size dispatch thread pool and should not
perform blocking IO, including core.async blocking ops (those that
end in !!).

Returns true unless port is already closed.

    
    
    Source
  


reduce

function
Usage: (reduce f init ch)
f should be a function of 2 arguments. Returns a channel containing
the single result of applying f to init and the first item from the
channel, then applying f to that result and the 2nd item, etc. If
the channel closes without yielding items, returns init and f is not
called. ch must close before reduce produces a result.

    
    
    Source
  


sliding-buffer

function
Usage: (sliding-buffer n)
Returns a buffer of size n. When full, puts will complete, and be
buffered, but oldest elements in buffer will be dropped (not
transferred).

    
    
    Source
  


solo-mode

function
Usage: (solo-mode mix mode)
Sets the solo mode of the mix. mode must be one of :mute or :pause

    
    
    Source
  


split

function
Usage: (split p ch)
       (split p ch t-buf-or-n f-buf-or-n)
Takes a predicate and a source channel and returns a vector of two
channels, the first of which will contain the values for which the
predicate returned true, the second those for which it returned
false.

The out channels will be unbuffered by default, or two buf-or-ns can
be supplied. The channels will close after the source channel has
closed.

    
    
    Source
  


sub

function
Usage: (sub p topic ch)
       (sub p topic ch close?)
Subscribes a channel to a topic of a pub.

By default the channel will be closed when the source closes,
but can be determined by the close? parameter.

    
    
    Source
  


take

function
Usage: (take n ch)
       (take n ch buf-or-n)
Returns a channel that will return, at most, n items from ch. After n items
 have been returned, or ch has been closed, the return channel will close.

The output channel is unbuffered by default, unless buf-or-n is given.

    
    
    Source
  


take!

function
Usage: (take! port fn1)
       (take! port fn1 on-caller?)
Asynchronously takes a val from port, passing to fn1. Will pass nil
if closed. If on-caller? (default true) is true, and value is
immediately available, will call fn1 on calling thread.

fn1 may be run in a fixed-size dispatch thread pool and should not
perform blocking IO, including core.async blocking ops (those that
end in !!).

Returns nil.

    
    
    Source
  


tap

function
Usage: (tap mult ch)
       (tap mult ch close?)
Copies the mult source onto the supplied channel.

By default the channel will be closed when the source closes,
but can be determined by the close? parameter.

    
    
    Source
  


thread

macro
Usage: (thread & body)
Executes the body in another thread, returning immediately to the
calling thread. Returns a channel which will receive the result of
the body when completed, then close.

    
    
    Source
  


thread-call

function
Usage: (thread-call f)
Executes f in another thread, returning immediately to the calling
thread. Returns a channel which will receive the result of calling
f when completed, then close.

    
    
    Source
  


timeout

function
Usage: (timeout msecs)
Returns a channel that will close after msecs

    
    
    Source
  


to-chan

function
Usage: (to-chan coll)
Deprecated - use to-chan! or to-chan!!

    
    Deprecated since core.async version 1.2
Source


to-chan!

function
Usage: (to-chan! coll)
Creates and returns a channel which contains the contents of coll,
closing when exhausted.

If accessing coll might block, use to-chan!! instead

    
    
    Source
  


to-chan!!

function
Usage: (to-chan!! coll)
Like to-chan! for use when accessing coll might block,
e.g. a lazy seq of blocking operations

    
    
    Source
  


toggle

function
Usage: (toggle mix state-map)
Atomically sets the state(s) of one or more channels in a mix. The
state map is a map of channels -> channel-state-map. A
channel-state-map is a map of attrs -> boolean, where attr is one or
more of :mute, :pause or :solo. Any states supplied are merged with
the current state.

Note that channels can be added to a mix via toggle, which can be
used to add channels in a particular (e.g. paused) state.

    
    
    Source
  


transduce

function
Usage: (transduce xform f init ch)
async/reduces a channel with a transformation (xform f).
Returns a channel containing the result.  ch must close before
transduce produces a result.

    
    
    Source
  


unblocking-buffer?

function
Usage: (unblocking-buffer? buff)
Returns true if a channel created with buff will never block. That is to say,
puts into this buffer will never cause the buffer to be full. 

    
    
    Source
  


unmix

function
Usage: (unmix mix ch)
Removes ch as an input to the mix

    
    
    Source
  


unmix-all

function
Usage: (unmix-all mix)
removes all inputs from the mix

    
    
    Source
  


unsub

function
Usage: (unsub p topic ch)
Unsubscribes a channel from a topic of a pub

    
    
    Source
  


unsub-all

function
Usage: (unsub-all p)
       (unsub-all p topic)
Unsubscribes all channels from a pub, or a topic of a pub

    
    
    Source
  


untap

function
Usage: (untap mult ch)
Disconnects a target channel from a mult

    
    
    Source
  


untap-all

function
Usage: (untap-all mult)
Disconnects all target channels from a mult

    
    
    Source
  

clojure.core.async.flow

A library for building concurrent, event driven data processing
flows out of communication-free functions, while centralizing
control, reporting, execution and error handling. Built on core.async.

The top-level construct is the flow, comprising:
a set of processes (generally, threads) - concurrent activities
a set of channels flowing data into and out of the processes
a set of channels for centralized control, reporting, error-handling,
  and execution of the processes

A flow is constructed from flow definition data which defines a
directed graph of processes and the connections between
them. Processes describe their I/O requirements and the
flow (library) itself creates channels and passes them to the
processes that requested them. See 'create-flow' for the
details. The flow definition provides a centralized place for policy
decisions regarding configuration, threading, buffering etc. 

It is expected that applications will rarely define processes but
instead use the API functions here, 'process' and 'step-process',
that implement the process protocol in terms of calls to ordinary
functions that include no communication or core.async code. In this
way the library helps you achieve a strict separation of your
application logic from its execution, communication, lifecycle and
monitoring.

Note that at several points the library calls upon the user to
supply ids for processes, inputs, outputs etc. These should be
keywords. When a namespaced keyword is required it is explicitly
stated. This documentation refers to various keywords utilized by
the library itself as ::flow/xyz, where ::flow is an alias for
clojure.core.async.flow

A process is represented in the flow definition by an implementation
of spi/ProcLauncher that starts it. See the spi docs for
details.


clojure.core.async.flow.impl.graph





Protocols



Graph

Protocol

    Known implementations: 
    

command-proc

function
Usage: (command-proc g pid cmd-id more-kvs)
synchronously sends a process-specific command with the given id 
and additional kvs to the process

      
      
      
    

inject

function
Usage: (inject g [pid io-id] msgs)
synchronously puts the messages on the channel 
corresponding to the input or output of the process

      
      
      
    

pause

function
Usage: (pause g)
pauses a running graph

      
      
      
    

pause-proc

function
Usage: (pause-proc g pid)
pauses a process

      
      
      
    

ping

function
Usage: (ping g)
pings all processes, which will put their status and state on the report channel

      
      
      
    

ping-proc

function
Usage: (ping-proc g pid)
pings the process, which will put its status and state on the report channel

      
      
      
    

resume

function
Usage: (resume g)
resumes a paused graph

      
      
      
    

resume-proc

function
Usage: (resume-proc g pid)
resumes a process

      
      
      
    

start

function
Usage: (start g)
starts the entire graph from init values. The processes start paused. 
Call resume-all or resume-proc to start flow.
returns {:report-chan -  a core.async chan for reading 
         :error-chan - a core.async chan for reading}

      
      
      
    

stop

function
Usage: (stop g)
shuts down the graph, stopping all procs, can be started again

      
      
      
    
Source

clojure.core.async.flow.spi





Protocols



ProcLauncher

Protocol
Note - definine a ProcLauncher is an advanced feature and should not
be needed for ordinary use of the library. This protocol is for
creating new types of Processes that are not possible to create
with ::flow/process.

A ProcLauncher is a constructor for a process, a thread of activity.
It has two functions - to describe the parameters and input/output
requirements of the process and to start it. The launcher should
acquire no resources, nor retain any connection to the started
process. A launcher may be called upon to start a process more than
once, and should start a new process each time start is called.

The process launched process must obey the following:

It must have 2 logical statuses, :paused and :running. In
the :paused status operation is suspended and no output is
produced.

When the process starts it must be :paused

Whenever it is reading or writing to any channel a process must use
alts!! and include a read of the ::flow/control channel, giving it
priority.

Command messages sent over the ::flow/control channel have the keys:
::flow/to - either ::flow/all or a process id
::flow/command - ::flow/stop|pause|resume|ping or process-specific

It must act upon any, and only, control messages whose ::flow/to key is its pid or ::flow/all
It must act upon the following values of ::flow/command:

::flow/stop - all resources should be cleaned up and any thread(s)
   should exit ordinarily - there will be no more subsequent use
   of the process.
::flow/pause - enter the :paused status
::flow/resume - enter the :running status and resume processing
::flow/ping - emit a ping message (format TBD) to
   the ::flow/report channel containing at least its pid and status

A process can define and respond to other commands in its own namespace.

A process should not transmit channel objects (use [pid io-id] data
coordinates instead) A process should not close channels

Finally, if a process encounters an error it must report it on the
::flow/error channel (format TBD) and attempt to continue, though it
may subsequently get a ::flow/stop command it must respect
Known implementations:

describe

function
Usage: (describe p)
returns a map with keys - :params, :ins and :outs,
each of which in turn is a map of keyword to docstring

:params describes the initial arguments to setup the state for the process
:ins enumerates the input[s], for which the graph will create channels
:outs enumerates the output[s], for which the graph may create channels.

describe may be called by users to understand how to use the
proc. It will also be called by the impl in order to discover what
channels are needed.

      
      
      
    

start

function
Usage: (start p {:keys [pid args ins outs resolver]})
return ignored, called for the
effect of starting the process (typically, starting its thread)

where:

:pid - the id of the process in the graph, so that e.g. it can refer to itself in control, reporting etc
:args - a map of param->val,  as supplied in the graph def
:ins - a map of in-id->readable-channel, plus the ::flow/control channel
:outs - a map of out-id->writeable-channel, plus the ::flow/error and ::flow/report channels
        N.B. outputs may be nil if not connected
:resolver - an impl of spi/Resolver, which can be used to find
            channels given their logical [pid cid] coordinates, as well as to
            obtain ExecutorServices corresponding to the
            logical :mixed/:io/:compute contexts

      
      
      
    
Source


Resolver

Protocol

    Known implementations: 
    

get-exec

function
Usage: (get-exec _ context)
returns the ExecutorService for the given context, one
of :mixed, :io, :compute

      
      
      
    

get-write-chan

function
Usage: (get-write-chan _ coord)
Given a tuple of [pid cid], returns a core.async chan to
write to or nil (in which case the output should be dropped,
e.g. nothing is connected).

      
      
      
    
Source
Logo & site design by Tom Hickey.
Clojure auto-documentation system by Tom Faulhaber.