core.async Documentation

The core.async.flow library is designed to separate your application's domain logic from its execution topology (threading, communication, lifecycle, and error handling).

Instead of manually wiring channels and managing threads, you use two primary abstractions:

1. Process: A thread of activity. You provide the computational logic, and the library handles the I/O, threading, and lifecycle.
2. Flow: A directed graph of processes communicating via channels. The entire topology is described by a single data structure.

By using this model, your core logic remains pure and testable, while the c.a.flow engine manages the complex 'plumbing' of the system.

The go macro creates an Inversion of Control (IOC) 'thread'. It transforms the body into a state machine. When the code reaches a blocking channel operation, the state machine is 'parked', and the underlying thread is released to do other work. When the operation completes, the code resumes.

Key operations inside go blocks:

• >! (put): Asynchronously puts a value into a channel.
• <! (take): Asynchronously takes a value from a channel.

Note: A go block returns a channel that will eventually contain the result of the last expression in the block.

The go macro executes its body in a special pool of lightweight threads. Unlike ordinary threads, channel operations inside a go block do not block the underlying thread; instead, they pause the execution of the go block itself (Inversion of Control).

Inside go blocks, use the non-blocking operators:

• >!: Put.
• <!: Take.

(let [c (a/chan)]
  (a/go (>! c "hello"))
  (assert (= "hello" (<!! (a/go (<! c)))))
  (a/close! c))

A promise channel is a special channel designed to accept exactly one value.

• Once a value is put into a promise channel, all current and future consumers will receive that same value.
• Subsequent put operations will complete but the values will be dropped.
• If the channel is closed without a value being put, consumers will receive nil forever.

core.async uses two primary models for running computations:

Go Blocks

go blocks represent lightweight processes. They can be "parked" (paused) by parking operations (>!, <!, alt!, alts!) without consuming a system thread. When the operation can complete, the block resumes. CRITICAL: Never use blocking operations (like <!!) or blocking I/O inside a go block. This can exhaust the thread pool and deadlock the entire system.

Threads

thread and thread-call execute processes in separate, unmanaged threads (similar to future). Because these are not multiplexed over a small pool, they are the correct place to perform blocking I/O or use blocking core.async operations (<!!, >!!).

You can add core.async to your project using various Clojure dependency management tools. The current latest version is 1.9.865.

### deps.edn
```clj
org.clojure/core.async {:mvn/version "1.9.865"}

Leiningen

[org.clojure/core.async "1.9.865"]

Maven

<dependency>
  <groupId>org.clojure</groupId>
  <artifactId>core.async</artifactId>
  <version>1.9.865</version>
</dependency>

To use core.async, ensure you are using Clojure 1.10.0 or higher. Add the following dependency to your project configuration:

{:deps
 {org.clojure/clojure {:mvn/version "1.12.0"}
  org.clojure/core.async {:mvn/version "1.8.741"}}}

To use the library in your code, require the clojure.core.async namespace. It is common to alias it as a and pull in the core operators (<!!, >!!, <!, >!) for convenience:

(ns my.ns
  (:require [clojure.core.async :as a :refer [<!! >!! <! >!]]))

In standard Clojure threads, you use blocking operations to interact with channels:

• >!!: Blocking put.
• <!!: Blocking take.

Warning: Using these on an unbuffered channel in the main thread will block the entire thread. To avoid this, use (a/thread ...) to run blocking operations in a background pool thread.

;; Blocking operations in a standard thread
(let [c (a/chan 10)]
  (>!! c "hello")
  (assert (= "hello" (<!! c)))
  (a/close! c))

;; Using a background thread to prevent blocking the main thread
(let [c (a/chan)]
  (a/thread (>!! c "hello"))
  (assert (= "hello" (<!! c)))
  (a/close! c))

A flow definition is a map passed to create-flow. It connects process launchers via connections.

Structure

• :procs: A map of pid -> proc-def.
  • proc-def contains :proc (the launcher), :args (passed to init), and optional :chan-opts.
• :conns: A collection of connection tuples: [[[from-pid out-id] [to-pid in-id]] ...].
  • If an output is connected multiple times, every connection receives the message (via core.async/mult).

Example Flow Definition

This example connects a :source-proc output to a :sink-proc input, passing external channels through the :args.

{:procs {:source-proc {:proc (process #'source-fn) 
                       :args {:source-chan in-chan}} 
         :sink-proc   {:proc (process #'sink-fn) 
                       :args {:sink-chan out-chan}}} 
 :conns [ [[:source-proc :out] [:sink-proc :in]] ]}