# Async/Await (Incubating)

version 6.0.0-alpha-1

## 1. Introduction

Groovy provides native `async`/`await` support, enabling developers to write concurrent code in a sequential, readable style. Rather than dealing with callbacks, `CompletableFuture` chains, or manual thread management, you express concurrency with two constructs:

- `async { …​ }` — start a closure on a background thread, returning an `Awaitable`

- `await expr` — block until an asynchronous result is available

On JDK 21+, async tasks automatically leverage virtual threads for optimal scalability. On JDK 17–20, a cached thread pool is used as a fallback.

The examples throughout this guide use an online multiplayer card game as a running theme — dealing hands, racing for the fastest play, streaming cards, and managing tournament rounds.

## 2. Getting Started

### 2.1. Your first async/await

`async { …​ }` starts work immediately on a background thread and returns an `Awaitable`. Use `await` to collect the result:

    def deck = ['2♠', '3♥', 'K♦', 'A♣']
    def card = async { deck.shuffled()[0] }
    println "You drew: ${await card}"

### 2.2. Exception handling

`await` unwraps `CompletionException` and `ExecutionException` automatically. Standard `try`/`catch` works exactly as with synchronous code:

    def drawFromEmpty = async {
        throw new IllegalStateException('deck is empty')
    }
    try {
        await drawFromEmpty
    } catch (IllegalStateException e) {
        // Original exception — no CompletionException wrapper
        assert e.message == 'deck is empty'
    }

### 2.3. CompletableFuture interop

`await` works directly with `CompletableFuture`, `CompletionStage`, and `Future` from Java libraries:

    // await works with CompletableFuture from Java libraries
    def future = CompletableFuture.supplyAsync { 'A♠' }
    assert await(future) == 'A♠'

## 3. Running Tasks in Parallel

The real power of async/await appears when you need to run several tasks concurrently and coordinate their results.

### 3.1. Waiting for all: `Awaitable.all()`

Deal cards to multiple players at the same time and wait for all hands:

    // Deal cards to three players concurrently
    def deck = [*1..9,'J','Q','K','A'].collectMany { rank ->
        ['♠','♥','♦','♣'].collect { suit -> "$rank$suit" }
    }.shuffled()
    def index = new AtomicInteger()
    def draw5 = { -> def i = index.getAndAdd(5); deck[i..i+4] }

    def alice = async { draw5() }
    def bob   = async { draw5() }
    def carol = async { draw5() }

    def (a, b, c) = await Awaitable.all(alice, bob, carol)
    assert a.size() == 5 && b.size() == 5 && c.size() == 5

Multi-argument `await` is syntactic sugar for `Awaitable.all()`:

    def a = async { 1 }
    def b = async { 2 }
    def c = async { 3 }

    // Parenthesized multi-arg await (sugar for Awaitable.all):
    def results = await(a, b, c)
    assert results == [1, 2, 3]

### 3.2. Racing: `Awaitable.any()`

Returns the result of the first task to complete — useful for fallback patterns or latency-sensitive code:

    // Race two servers — use whichever responds first
    def primary  = async { Thread.sleep(200); 'primary-response' }
    def fallback = async { 'fallback-response' }

    def response = await Awaitable.any(primary, fallback)
    assert response == 'fallback-response'

### 3.3. First success: `Awaitable.first()`

Like JavaScript’s `Promise.any()` — returns the first successful result, silently ignoring individual failures. Only fails when every task fails:

    // Try multiple sources — use the first that succeeds
    def failing    = async { throw new RuntimeException('server down') }
    def succeeding = async { 'card-data-from-cache' }

    def result = await Awaitable.first(failing, succeeding)
    assert result == 'card-data-from-cache'

### 3.4. Inspecting all outcomes: `Awaitable.allSettled()`

Waits for all tasks to finish (succeed or fail) without throwing. Returns an `AwaitResult` list where each entry has `success`, `value`, and `error` fields:

    def save1 = async { 42 }
    def save2 = async { throw new RuntimeException('db error') }

    def results = await Awaitable.allSettled(save1, save2)
    assert results[0].success && results[0].value == 42
    assert !results[1].success && results[1].error.message == 'db error'

### 3.5. Combinator summary

Combinator | Completes when | On failure | Use case
--- | --- | --- | ---
`Awaitable.all` | All succeed | Fails immediately on first failure (fail-fast) | Gather results from independent tasks
`Awaitable.allSettled` | All complete (success or fail) | Never throws; failures captured in `AwaitResult` list | Inspect every outcome, e.g. partial-success reporting
`Awaitable.any` | First task completes (success or failure) | Propagates the first completion’s result or error | Latency-sensitive races, fastest-response wins
`Awaitable.first` | First task succeeds, or all fail | Throws only when every source fails (aggregate error) | Hedged requests, graceful degradation with fallbacks

## 4. Generators and Streaming

### 4.1. Producing values with `yield return`

An `async` closure containing `yield return` becomes a generator — it lazily produces a sequence of values. The generator runs on a background thread and blocks on each `yield return` until the consumer is ready, providing natural back-pressure:

    def dealCards = async {
        def suits = ['♠', '♥', '♦', '♣']
        def ranks = ['A', '2', '3', '4', '5', '6', '7', '8', '9', '10', 'J', 'Q', 'K']
        for (suit in suits) {
            for (rank in ranks) {
                yield return "$rank$suit"
            }
        }
    }
    def cards = dealCards.collect()
    assert cards.size() == 52
    assert cards.first() == 'A♠'
    assert cards.last() == 'K♣'

Generators return a standard `Iterable`, so regular `for` loops and Groovy collection methods (`collect`, `findAll`, `take`, etc.) work out of the box:

    def scores = async {
        for (s in [100, 250, 75]) { yield return s }
    }
    // Generators return Iterable — regular for and collect work
    assert scores.collect { it * 2 } == [200, 500, 150]

## 5. Channels

Channels provide Go-style inter-task communication. A producer sends values into a channel; a consumer receives them as they arrive. The channel handles synchronization and optional buffering — no shared mutable state needed:

    def cardStream = AsyncChannel.create(3)

    // Dealer — sends cards concurrently
    async {
        for (card in ['A♠', 'K♥', 'Q♦', 'J♣']) {
            await cardStream.send(card)
        }
        cardStream.close()
    }

    // Player — receives cards as they arrive
    def hand = []
    for await (card in cardStream) {
        hand << card
    }
    assert hand == ['A♠', 'K♥', 'Q♦', 'J♣']

Channels support unbuffered (rendezvous, `create()`) and buffered (`create(n)`) modes. Sending blocks when the buffer is full; receiving blocks when empty. With virtual threads, this blocking is essentially free.

Since channels implement `Iterable`, they also work with regular `for` loops and Groovy collection methods.

## 6. Channel Composition

`AsyncChannel` supports composable pipeline operations. Each composition method returns a new channel connected to the source by a background task.

### 6.1. filter

    def source = AsyncChannel.create(10)
    def evens = source.filter { it % 2 == 0 }

    async {
        (1..10).each { source.send(it) }
        source.close()
    }

    for await (val in evens) { println val }  // 2, 4, 6, 8, 10

### 6.2. map

    def doubled = source.map { it * 2 }

### 6.3. Chaining

Operations compose naturally:

    def pipeline = source
        .filter { it > 50 }
        .map { it * 2 }

    for await (val in pipeline) { process(val) }

### 6.4. merge

Combines two channels into one, interleaving values as they arrive:

    def merged = channelA.merge(channelB)
    for await (event in merged) { handle(event) }

### 6.5. split

Partitions a channel into two based on a predicate:

    def (highPriority, normal) = orders.split { it.priority > 5 }

### 6.6. tap

Sends a copy of each value to a monitoring channel while passing through unchanged:

    def monitor = AsyncChannel.create(100)
    def output = source.tap(monitor)
    // output gets all values; monitor gets copies for logging

## 7. Channel Select

`ChannelSelect` waits for the first available value from multiple channels — like Go’s `select` statement:

    import groovy.concurrent.ChannelSelect

    def prices = AsyncChannel.create(10)
    def alerts = AsyncChannel.create(10)

    def sel = ChannelSelect.from(prices, alerts)
    def result = await sel.select()
    println "Channel ${result.index}: ${result.value}"

## 8. BroadcastChannel

A `BroadcastChannel` delivers every value to all subscribers (one-to-many), unlike `AsyncChannel` which is point-to-point:

    import groovy.concurrent.BroadcastChannel

    def broadcast = BroadcastChannel.create()
    def sub1 = broadcast.subscribe()
    def sub2 = broadcast.subscribe()

    async {
        broadcast.send('hello')
        broadcast.send('world')
        broadcast.close()
    }

    // Both subscribers receive both messages
    for await (msg in sub1) { println "Sub1: $msg" }
    for await (msg in sub2) { println "Sub2: $msg" }

Late subscribers only receive messages sent after they subscribe.

### 8.1. Flow.Publisher interop

A `BroadcastChannel` can also be exposed as a `java.util.concurrent.Flow.Publisher` via `asPublisher()`. Each call to `subscribe(Flow.Subscriber)` on the returned publisher creates a new per-subscriber `AsyncChannel` under the hood and respects standard Reactive Streams demand (`request(n)`):

    def broadcast = BroadcastChannel.create()
    def publisher = broadcast.asPublisher()

    // Compose with any Reactive Streams operator, or consume with `for await`:
    for await (msg in publisher) {
        println "Got: $msg"
    }

| Backpressure policy. The publisher bridge uses lossless, sender-gated backpressure: `send()` awaits delivery to every live subscriber, and each per-subscriber channel has a bounded buffer (default 16). A subscriber that never calls `request(n)` — or requests slowly — will fill its buffer, after which `broadcast.send(…​)` stalls for all subscribers until that subscriber catches up. In other words, the slowest subscriber controls producer throughput. This matches the point-to-point semantics of `subscribe()` (values are neither dropped nor reordered). If you need decoupled per-subscriber policies (drop-newest, drop-oldest, latest-only, or unbounded buffering), wrap the publisher with a Reactive Streams operator of your choice, or use a subscriber that drains promptly with `request(Long.MAX_VALUE)`.

| Terminal signals. Per Reactive Streams §1.7, terminal completion is not gated by demand: once the underlying `BroadcastChannel` is closed and the per-subscriber buffer is drained, the subscriber receives `onComplete` even if it has no outstanding `request(n)`. Likewise, `onError` for a spec-violating `request(n ⇐ 0)` (§3.9) is signalled through the same single-threaded delivery path as `onNext` / `onComplete`, preserving §1.3 signal serialisation; subscribers do not see those signals racing on the caller’s thread.

## 9. Deferred Cleanup with `defer`

The `defer` keyword schedules a cleanup action to run when the enclosing `async` closure completes, regardless of success or failure. Multiple deferred actions execute in LIFO order — last registered, first to run — making it natural to pair resource acquisition with cleanup:

    def log = []
    def task = async {
        log << 'open connection'
        defer { log << 'close connection' }
        log << 'open transaction'
        defer { log << 'close transaction' }
        log << 'save game state'
        'saved'
    }
    assert await(task) == 'saved'
    // Deferred actions run in LIFO order — last registered, first to run
    assert log == ['open connection', 'open transaction', 'save game state',
                   'close transaction', 'close connection']

Deferred actions always run, even when an exception occurs:

    def cleaned = false
    def task = async {
        defer { cleaned = true }
        throw new RuntimeException('save failed')
    }
    try {
        await task
    } catch (RuntimeException e) {
        assert e.message == 'save failed'
    }
    // Deferred actions run even when an exception occurs
    assert cleaned

If a deferred action returns an `Awaitable` or `Future`, the result is awaited before the next deferred action runs, ensuring orderly cleanup of asynchronous resources.

## 10. Structured Concurrency

Structured concurrency ensures that concurrent tasks have clear ownership and bounded lifetimes — no orphaned background work, no silent failures leaking across your application. This idea is gaining momentum across the industry (Java’s JEP 453, Go’s `errgroup`, Kotlin’s coroutine scopes) because it makes concurrent code easier to reason about, test, and debug. Groovy’s `AsyncScope` provides these guarantees today, even on JDK versions before Project Loom’s `StructuredTaskScope` ships as a final API.

`AsyncScope` binds the lifetime of child tasks to a scope. When the scope exits, all children are guaranteed complete (or cancelled). This prevents orphaned tasks and silent failures:

    // Run a tournament round — all tables play concurrently
    def results = AsyncScope.withScope { scope ->
        def table1 = scope.async { [winner: 'Alice',  score: 320] }
        def table2 = scope.async { [winner: 'Bob',    score: 280] }
        def table3 = scope.async { [winner: 'Carol',  score: 410] }
        [await(table1), await(table2), await(table3)]
    }
    // All tables guaranteed complete when withScope returns
    assert results.size() == 3
    assert results.max { it.score }.winner == 'Carol'

By default, the scope uses fail-fast semantics — if any child fails, all siblings are cancelled immediately:

    try {
        AsyncScope.withScope { scope ->
            scope.async { Thread.sleep(5000); 'still playing' }
            scope.async { throw new RuntimeException('player disconnected') }
        }
    } catch (RuntimeException e) {
        // First failure cancels all siblings and propagates
        assert e.message == 'player disconnected'
    }

The scope waits for every child to finish, even without explicit `await` calls:

    def completed = new AtomicInteger(0)
    AsyncScope.withScope { scope ->
        3.times { scope.async { Thread.sleep(50); completed.incrementAndGet() } }
    }
    // All children have completed — even without explicit await
    assert completed.get() == 3

On JDK 25+, scope tracking uses `ScopedValue` for optimal virtual thread performance (no per-thread storage, automatic inheritance). On JDK 17–24, a `ThreadLocal` fallback is used transparently.

## 11. Advanced Topics

### 11.1. Consuming with `for await`

`for await` iterates over any async source. For generators and plain collections, it works identically to a regular `for` loop:

    def topCards = async {
        for (card in ['A♠', 'K♥', 'Q♦']) {
            yield return card
        }
    }
    def hand = []
    for await (card in topCards) {
        hand << card
    }
    assert hand == ['A♠', 'K♥', 'Q♦']

The key value of `for await` is with reactive types (Reactor `Flux`, RxJava `Observable`, `java.util.concurrent.Flow.Publisher`) where it automatically converts the source to a blocking iterable via the adapter SPI. Without `for await`, you would need to call the conversion manually (e.g., `flux.toIterable()`). For generators and plain collections, a regular `for` loop works identically.

The JDK’s `Flow.Publisher` is supported out of the box — no adapter module needed. This means streams exposed by built-in sources such as `Agent.changes()` (see Agent) and `BroadcastChannel.asPublisher()` (see BroadcastChannel), or any third-party publisher using the JDK interfaces, compose directly with `for await` and `await`:

    import groovy.concurrent.Agent

    def agent = Agent.create(0)
    try {
        async {
            3.times { agent.send { it + 1 } }
            agent.shutdown()
        }
        for await (v in agent.changes()) {
            println "value is now $v"
        }
    } finally {
        agent.shutdown()
    }

### 11.2. Timeouts and Delays

Apply a deadline to any task. If it doesn’t complete in time, a `TimeoutException` is thrown:

    def slowPlayer = async { Thread.sleep(5000); 'finally played' }
    try {
        await Awaitable.orTimeoutMillis(slowPlayer, 100)
    } catch (TimeoutException e) {
        // Player took too long — turn forfeited
        assert true
    }

Or use a fallback value instead of throwing:

    def slowPlayer = async { Thread.sleep(5000); 'deliberate move' }
    def move = await Awaitable.completeOnTimeoutMillis(slowPlayer, 'auto-pass', 100)
    assert move == 'auto-pass'

`Awaitable.delay()` pauses without blocking a thread:

    long start = System.currentTimeMillis()
    await Awaitable.delay(100)   // pause without blocking a thread
    assert System.currentTimeMillis() - start >= 90

### 11.3. Framework Adapters

`await` natively understands `Awaitable`, `CompletableFuture`, `CompletionStage`, and `Future`. `for await` natively understands `java.util.concurrent.Flow.Publisher` in addition to regular `Iterable` sources. Third-party async types can be supported by implementing the `AwaitableAdapter` SPI and registering it via `META-INF/services/groovy.concurrent.AwaitableAdapter`.

Drop-in adapter modules are provided for:

- `groovy-reactor` — `await` on `Mono`, `for await` over `Flux`

- `groovy-rxjava` — `await` on `Single`/`Maybe`/`Completable`, `for await` over `Observable`/`Flowable`

For example, without the adapter you must manually convert RxJava types:

    // Without groovy-rxjava — manual conversion
    def result = Single.just('hello').toCompletionStage().toCompletableFuture().join()

With `groovy-rxjava` on the classpath, the conversion is transparent:

    // With groovy-rxjava — adapter handles the plumbing
    def result = await Awaitable.from(Single.just('hello'))

### 11.4. Executor Configuration

By default, `async` uses:

- JDK 21+: a virtual-thread-per-task executor

- JDK 17–20: a cached daemon thread pool (max 256 threads, configurable via `groovy.async.parallelism` system property)

You can override the executor:

    import org.apache.groovy.runtime.async.AsyncSupport
    import java.util.concurrent.Executors

    AsyncSupport.setExecutor(Executors.newFixedThreadPool(4))
    AsyncSupport.resetExecutor()  // restore default

### 11.5. Integration with JDK Classes

`await` works with any JDK API that returns a `CompletableFuture`, `CompletionStage`, or `Future`. This means you can combine process execution, asynchronous file I/O, and HTTP calls in a single `async` block — all running concurrently:

    // Task 1: Run a process and await its completion
    def proc = async {
        def p = echoCmd.execute() // echo Hello from Groovy
        await p.onExit()
        p.text.trim()
    }

    // Task 2: Read a file asynchronously
    def fileContent = async {
        await etcPassword.textAsync
    }

    // Task 3: Fetch a web page using Groovy's HttpBuilder
    def webContent = async {
        def http = HttpBuilder.http { baseUri bankUrl }
        http.get('/withdraw/100').body
    }

    // All three tasks run concurrently — collect results
    var (echo, file, html) = await proc, fileContent, webContent
    assert echo == 'Hello from Groovy'
    assert file =~ /sEcrEt/
    assert html == '<html>SUCCESS</html>'

Task 3 uses Groovy’s `HttpBuilder` (from `groovy-http-builder`), which wraps JDK `HttpClient` with a concise DSL. For lower-level control, you can also use `HttpClient` directly:

    import java.net.http.HttpClient
    import java.net.http.HttpRequest
    import java.net.http.HttpResponse

    def webContent = async {
        def client = HttpClient.newHttpClient()
        def req = HttpRequest.newBuilder()
            .uri(URI.create('https://api.example.com/data'))
            .build()
        def resp = await client.sendAsync(req, HttpResponse.BodyHandlers.ofString())
        resp.body()
    }

Key JDK classes that work with `await` out of the box:

JDK class | Method returning async result
--- | ---
`Process` | `onExit()` → `CompletableFuture<Process>`
`HttpClient` | `sendAsync(…​)` → `CompletableFuture<HttpResponse<T>>`
`Path` (via `groovy-nio`) | `getTextAsync()`, `getBytesAsync()`, `writeAsync(…​)` → `CompletableFuture`
`AsynchronousFileChannel` | `read(…​)`, `write(…​)`, `lock(…​)` → `Future<Integer>` / `Future<FileLock>`
`CompletableFuture` | `supplyAsync(…​)`, `allOf(…​)`, `anyOf(…​)` → `CompletableFuture<T>`

### 11.6. Async Closure Factories

Because `async { …​ }` is just an expression that returns an `Awaitable`, you can wrap it in an ordinary method or closure to create a reusable factory:

    // A method that returns an async task — a simple factory
    def dealCard(List deck) {
        async { deck.shuffled()[0] }
    }

    def numPlayers = 3
    // each player gets a card from their own deck
    def cards = (1..numPlayers).collect { dealCard(['A♠', 'K♥', 'Q♦', 'J♣']) }

    def hands = await Awaitable.all(*cards)
    assert hands.size() == 3
    assert hands.every { it in ['A♠', 'K♥', 'Q♦', 'J♣'] }

This is a natural way to build reusable async building blocks without any special framework support.

## 12. Best Practices

### 12.1. Prefer returning values over mutating shared state

Async closures run on separate threads. Mutating shared variables from multiple closures is a race condition:

    // UNSAFE — shared mutation without synchronization
    var count = 0
    def tasks = (1..100).collect { async { count++ } }
    tasks.each { await it }
    // count may not be 100!

Instead, return values and collect results:

    // SAFE — no shared mutation
    def tasks = (1..100).collect { n -> async { n } }
    def results = await Awaitable.all(*tasks)
    assert results.sum() == 5050

When shared mutable state is unavoidable, use the appropriate concurrency-aware type, e.g. `AtomicInteger` for a shared counter, or thread-safe types from `java.util.concurrent` for players concurrently drawing cards from a shared deck.

### 12.2. Choosing the right tool

Feature | Use when…​
--- | ---
`async`/`await` | You have sequential steps involving I/O or blocking work and want code that reads top-to-bottom.
`Awaitable.all` / `any` / `first` | You need to launch independent tasks and collect all results, race them, or take the first success.
`yield return` / `for await` | You’re producing or consuming a stream of values — paginated APIs, card dealing, log tailing.
`defer` | You acquire resources at different points and want guaranteed cleanup without nested `try`/`finally`.
`AsyncChannel` | Two or more tasks need to communicate — producer/consumer, fan-out/fan-in, or hand-off.
`AsyncScope` | You want child task lifetimes tied to a scope with automatic cancellation on failure.
Framework adapters | You’re already using Reactor or RxJava and want `await` / `for await` to work with their types.

## 13. Quick Reference

Construct | Description
--- | ---
`async { …​ }` | Start a closure on a background thread. Returns `Awaitable` (or `Iterable` for generators).
`await expr` | Block until the result is available. Rethrows the original exception.
`await(a, b, c)` | Wait for all — syntactic sugar for `await Awaitable.all(a, b, c)`.
`yield return expr` | Produce a value from an async generator. Consumer blocks until ready.
`for await (x in src)` | Iterate over an async source (generator, channel, Flux, Observable, etc.).
`defer expr` | Schedule a cleanup action (LIFO order) inside an `async` closure.
`AsyncChannel.create(n)` | Create a buffered (or unbuffered) channel for inter-task communication.
`AsyncScope.withScope { …​ }` | Structured concurrency — all children complete (or are cancelled) on scope exit.
`Awaitable.orTimeoutMillis` | Apply a deadline. Throws `TimeoutException` if the task exceeds it.
`Awaitable.completeOnTimeoutMillis` | Apply a deadline with a fallback value instead of throwing.
`Awaitable.delay(ms)` | Non-blocking pause.

All keywords (`async`, `await`, `defer`) are contextual — they can still be used as variable or method names in existing code.