You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@groovy.apache.org by pa...@apache.org on 2026/04/03 05:59:17 UTC
(groovy) branch GROOVY_9381_5 updated (31bfac028b -> aed7e34a28)
This is an automated email from the ASF dual-hosted git repository.
paulk pushed a change to branch GROOVY_9381_5
in repository https://gitbox.apache.org/repos/asf/groovy.git
discard 31bfac028b GROOVY-9381: Add native async/await support
new aed7e34a28 GROOVY-9381: Add native async/await support
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (31bfac028b)
\
N -- N -- N refs/heads/GROOVY_9381_5 (aed7e34a28)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/spec/doc/core-async-await.adoc | 5 +++--
src/spec/test/AsyncAwaitSpecTest.groovy | 5 ++---
2 files changed, 5 insertions(+), 5 deletions(-)
(groovy) 01/01: GROOVY-9381: Add native async/await support
Posted by pa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
paulk pushed a commit to branch GROOVY_9381_5
in repository https://gitbox.apache.org/repos/asf/groovy.git
commit aed7e34a28ad70e1c2b074cf4e53ba5660427292
Author: Paul King <pa...@asert.com.au>
AuthorDate: Fri Apr 3 15:40:26 2026 +1000
GROOVY-9381: Add native async/await support
---
.../apache/groovy/parser/antlr4/AstBuilder.java | 4 +-
.../apache/groovy/runtime/async/AsyncSupport.java | 27 ++
.../groovy/transform/AsyncTransformHelper.java | 16 +-
src/spec/doc/core-async-await.adoc | 449 ++++++++++-----------
src/spec/test/AsyncAwaitSpecTest.groovy | 392 +++++++++---------
src/test/groovy/groovy/AsyncAwaitTest.groovy | 175 +++++---
6 files changed, 578 insertions(+), 485 deletions(-)
diff --git a/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java b/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java
index 40b1c55c89..af40395cca 100644
--- a/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java
+++ b/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java
@@ -2971,10 +2971,10 @@ public class AstBuilder extends GroovyParserBaseVisitor<Object> {
ClosureExpression genClosure = new ClosureExpression(newParams, closure.getCode());
genClosure.setVariableScope(closure.getVariableScope());
genClosure.setSourcePosition(closure);
- return configureAST(AsyncTransformHelper.buildWrapAsyncGeneratorCall(
+ return configureAST(AsyncTransformHelper.buildAsyncGeneratorCall(
new ArgumentListExpression(genClosure)), ctx);
} else {
- return configureAST(AsyncTransformHelper.buildWrapAsyncCall(
+ return configureAST(AsyncTransformHelper.buildAsyncCall(
new ArgumentListExpression(closure)), ctx);
}
}
diff --git a/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java b/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java
index 52d3f3837e..0f6d1d4f8b 100644
--- a/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java
+++ b/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java
@@ -328,6 +328,33 @@ public class AsyncSupport {
};
}
+ /**
+ * Starts a generator immediately, returning an {@link Iterable} backed by a
+ * {@link GeneratorBridge}. This is the runtime entry point for
+ * {@code async { ... yield return ... }} expressions.
+ *
+ * @param closure the generator closure; receives a GeneratorBridge as first parameter
+ * @param <T> the element type
+ * @return an Iterable that yields values from the generator
+ */
+ @SuppressWarnings("unchecked")
+ public static <T> Iterable<T> asyncGenerator(Closure<?> closure) {
+ Objects.requireNonNull(closure, "closure must not be null");
+ GeneratorBridge<T> bridge = new GeneratorBridge<>();
+ Object[] args = new Object[]{bridge};
+ defaultExecutor.execute(() -> {
+ try {
+ closure.call(args);
+ bridge.complete();
+ } catch (GeneratorBridge.GeneratorClosedException ignored) {
+ // Consumer closed early — normal for break in for-await
+ } catch (Throwable t) {
+ bridge.completeExceptionally(t);
+ }
+ });
+ return () -> bridge;
+ }
+
// ---- for-await (blocking iterable conversion) -----------------------
/**
diff --git a/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java b/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java
index 00b323e0a6..766fe72d78 100644
--- a/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java
+++ b/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java
@@ -57,8 +57,8 @@ public final class AsyncTransformHelper {
private static final String ASYNC_GEN_PARAM_NAME = "$__asyncGen__";
private static final String AWAIT_METHOD = "await";
private static final String YIELD_RETURN_METHOD = "yieldReturn";
- private static final String WRAP_ASYNC_METHOD = "wrapAsync";
- private static final String WRAP_ASYNC_GENERATOR_METHOD = "wrapAsyncGenerator";
+ private static final String ASYNC_METHOD = "async";
+ private static final String ASYNC_GENERATOR_METHOD = "asyncGenerator";
private static final String TO_BLOCKING_ITERABLE_METHOD = "toBlockingIterable";
private static final String CLOSE_ITERABLE_METHOD = "closeIterable";
private static final String CREATE_DEFER_SCOPE_METHOD = "createDeferScope";
@@ -100,17 +100,17 @@ public final class AsyncTransformHelper {
}
/**
- * Builds {@code AsyncSupport.wrapAsync(closure)}.
+ * Builds {@code AsyncSupport.async(closure)} — starts immediately, returns Awaitable.
*/
- public static Expression buildWrapAsyncCall(Expression closure) {
- return callX(ASYNC_SUPPORT_TYPE, WRAP_ASYNC_METHOD, ensureArgs(closure));
+ public static Expression buildAsyncCall(Expression closure) {
+ return callX(ASYNC_SUPPORT_TYPE, ASYNC_METHOD, ensureArgs(closure));
}
/**
- * Builds {@code AsyncSupport.wrapAsyncGenerator(closure)}.
+ * Builds {@code AsyncSupport.asyncGenerator(closure)} — starts immediately, returns Iterable.
*/
- public static Expression buildWrapAsyncGeneratorCall(Expression closure) {
- return callX(ASYNC_SUPPORT_TYPE, WRAP_ASYNC_GENERATOR_METHOD, ensureArgs(closure));
+ public static Expression buildAsyncGeneratorCall(Expression closure) {
+ return callX(ASYNC_SUPPORT_TYPE, ASYNC_GENERATOR_METHOD, ensureArgs(closure));
}
/**
diff --git a/src/spec/doc/core-async-await.adoc b/src/spec/doc/core-async-await.adoc
index b4d609cb75..afa0b298c8 100644
--- a/src/spec/doc/core-async-await.adoc
+++ b/src/spec/doc/core-async-await.adoc
@@ -21,134 +21,72 @@
= Async/Await (Incubating)
-[[async-await]]
+[[async-intro]]
== 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 simple constructs:
+concurrency with two constructs:
-* **`async { ... }`** — execute a closure asynchronously, returning an `Awaitable`
+* **`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
https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/lang/Thread.html#ofVirtual()[virtual threads]
for optimal scalability. On JDK 17–20, a cached thread pool is used as a
-fallback, providing correct behavior with slightly higher resource cost for
-high-concurrency scenarios.
+fallback.
-A `defer` statement for cleanup is also provided, inspired by Go's `defer` keyword.
+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.
-[[async-basic]]
-== Basic Usage
+[[async-getting-started]]
+== Getting Started
-The simplest use of `async`/`await` spawns work in a background thread and
-collects the result:
+=== Your first async/await
-[source,groovy]
-----
-include::../test/AsyncAwaitSpecTest.groovy[tags=basic_async_await,indent=0]
-----
-
-`async { ... }` creates a _reusable async closure_ — the work does **not**
-start immediately. You must call it with `()` to launch the computation and
-get back an `Awaitable`:
-
-[source,groovy]
-----
-def task = async { expensiveWork() } // defines the async closure — nothing runs yet
-def awaitable = task() // () starts the work, returns an Awaitable
-def result = await awaitable // blocks until the result is ready
-----
-
-Each `()` invocation starts a fresh, independent execution, so the same
-async closure can be reused. If you want work to start immediately without
-the extra `()` call, use `Awaitable.go`:
+`async { ... }` starts work immediately on a background thread and returns an
+`Awaitable`. Use `await` to collect the result:
[source,groovy]
----
-def task = Awaitable.go { expensiveWork() } // starts immediately
-def result = await task // no () needed
+include::../test/AsyncAwaitSpecTest.groovy[tags=draw_card,indent=0]
----
-Here is a more realistic example — a sequential workflow where each step
-depends on the previous one:
+=== Exception handling
-[source,groovy]
-----
-include::../test/AsyncAwaitSpecTest.groovy[tags=sequential_workflow,indent=0]
-----
-
-The code reads top-to-bottom like synchronous code, but the entire workflow
-runs in a background thread. No callbacks, no `.thenCompose()` chains.
-
-[[async-shared-state]]
-=== A Note on Shared State
-
-Async closures run on a separate thread, so mutating shared variables from
-multiple async closures is a race condition — just as it would be with
-`Thread.start { ... }`:
-
-[source,groovy]
-----
-// UNSAFE — count is shared and mutated without synchronization
-var count = 0
-def tasks = (1..100).collect { async { count++ } }
-tasks.each { await it() }
-// count may not be 100!
-----
-
-Prefer returning values from async closures and collecting results with
-`await` or combinators, rather than mutating shared state:
-
-[source,groovy]
-----
-// SAFE — each async closure returns a value, no shared mutation
-def tasks = (1..100).collect { n -> async { n } }
-def results = await(tasks.collect { it() })
-assert results.sum() == 5050
-----
-
-When shared mutable state is unavoidable, use `AtomicInteger`,
-`ConcurrentHashMap`, or other thread-safe types from `java.util.concurrent`.
-
-[[async-exception-handling]]
-=== Exception Handling
-
-`await` automatically unwraps `CompletionException` and `ExecutionException`,
-rethrowing the original exception. Standard `try`/`catch` works naturally:
+`await` unwraps `CompletionException` and `ExecutionException` automatically.
+Standard `try`/`catch` works exactly as with synchronous code:
[source,groovy]
----
include::../test/AsyncAwaitSpecTest.groovy[tags=exception_handling,indent=0]
----
-[[async-cf-interop]]
-=== CompletableFuture Interop
+=== CompletableFuture interop
`await` works directly with `CompletableFuture`, `CompletionStage`, and `Future`
-from Java libraries — no wrapping needed:
+from Java libraries:
[source,groovy]
----
include::../test/AsyncAwaitSpecTest.groovy[tags=cf_interop,indent=0]
----
-[[async-combinators]]
-== Combinators
+[[async-parallel]]
+== Running Tasks in Parallel
-`Awaitable` provides static combinator methods for coordinating multiple
-concurrent tasks.
+The real power of async/await appears when you need to run several tasks
+concurrently and coordinate their results.
-[[async-all]]
-=== Waiting for All: `Awaitable.all()` and Multi-arg `await`
+=== Waiting for all: `Awaitable.all()`
-Launch several tasks in parallel and wait for all results:
+Deal cards to multiple players at the same time and wait for all hands:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=parallel_tasks,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=deal_hands,indent=0]
----
Multi-argument `await` is syntactic sugar for `Awaitable.all()`:
@@ -158,279 +96,324 @@ Multi-argument `await` is syntactic sugar for `Awaitable.all()`:
include::../test/AsyncAwaitSpecTest.groovy[tags=multi_arg_await,indent=0]
----
-[[async-any]]
=== Racing: `Awaitable.any()`
-Returns the result of the first task to complete (success or failure):
+Returns the result of the first task to complete — useful for fallback
+patterns or latency-sensitive code:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=racing_tasks,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=fastest_server,indent=0]
----
-[[async-first]]
-=== First Success: `Awaitable.first()`
+=== First success: `Awaitable.first()`
Like JavaScript's `Promise.any()` — returns the first _successful_ result,
-ignoring individual failures. Only fails when all tasks fail:
+silently ignoring individual failures. Only fails when every task fails:
[source,groovy]
----
include::../test/AsyncAwaitSpecTest.groovy[tags=first_success,indent=0]
----
-Typical use cases include hedged requests (send to multiple endpoints, use
-whichever responds first) and graceful degradation (try primary, then fallback).
-
-[[async-all-settled]]
-=== Inspecting All Outcomes: `Awaitable.allSettled()`
+=== Inspecting all outcomes: `Awaitable.allSettled()`
-Waits for all tasks to complete (succeed or fail) without throwing, returning
-an `AwaitResult` list:
+Waits for all tasks to finish (succeed or fail) without throwing. Returns an
+`AwaitResult` list where each entry has `success`, `value`, and `error` fields:
[source,groovy]
----
include::../test/AsyncAwaitSpecTest.groovy[tags=all_settled,indent=0]
----
-Each `AwaitResult` has `success` (boolean), `value` (if successful), and
-`error` (if failed).
+[[async-generators]]
+== Generators and Streaming
-[[async-go]]
-=== Lightweight Spawn: `Awaitable.go()`
+=== Producing values with `yield return`
-A shorthand for spawning a task on the default executor:
+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:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=go_spawn,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=deck_generator,indent=0]
----
-[[async-structured]]
-== Structured Concurrency
-
-Structured concurrency binds the lifetime of concurrent tasks to a scope.
-When the scope exits, all child tasks are guaranteed to have completed (or
-been cancelled). This prevents orphaned tasks, resource leaks, and silent
-failures from unobserved exceptions.
-
-Use `AsyncScope.withScope` to create a scope:
+Generators return a standard `Iterable`, so regular `for` loops and Groovy
+collection methods (`collect`, `findAll`, `take`, etc.) work out of the box:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=structured_concurrency,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=generator_regular_for,indent=0]
----
-[[async-fail-fast]]
-=== Fail-fast Cancellation
+=== Consuming with `for await`
-By default, if any child task fails, all siblings are cancelled immediately.
-The first failure propagates as the scope's exception:
+`for await` iterates over any async source. For generators and plain collections,
+it works identically to a regular `for` loop:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=scope_fail_fast,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=for_await_generator,indent=0]
----
-[[async-scope-waits]]
-=== All Children Complete
+The key value of `for await` is with **reactive types** (Reactor `Flux`, RxJava
+`Observable`) 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 scope blocks until every child task finishes, even if no explicit `await`
-is used:
+[[async-channels]]
+== 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:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=scope_waits,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=channel,indent=0]
----
-[[async-timeout]]
-== Timeouts and Delays
+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.
-[[async-delay]]
-=== Non-blocking Delay
+Since channels implement `Iterable`, they also work with regular `for` loops
+and Groovy collection methods.
-`Awaitable.delay()` pauses without blocking a thread:
+[[async-defer]]
+== 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:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=delay,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=defer_basic,indent=0]
----
-[[async-or-timeout]]
-=== Fail-fast Timeout
-
-Apply a deadline to any task. If the task doesn't complete in time,
-a `TimeoutException` is thrown:
+Deferred actions always run, even when an exception occurs:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=timeout,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=defer_exception,indent=0]
----
-[[async-complete-on-timeout]]
-=== Fallback on Timeout
+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.
-Use a fallback value instead of throwing:
+[[async-diving-deeper]]
+== Diving Deeper
+
+[[async-structured]]
+=== Structured Concurrency
+
+`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:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=timeout_fallback,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=structured_concurrency,indent=0]
----
-[[async-generators]]
-== Generators with `yield return`
-
-Async closures containing `yield return` become _generators_ — functions that
-lazily produce a sequence of values. The generator runs on a background thread
-and each `yield return` hands a value to the consumer, blocking until the
-consumer is ready for the next one (natural back-pressure):
+By default, the scope uses **fail-fast** semantics — if any child fails,
+all siblings are cancelled immediately:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=yield_return,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=scope_fail_fast,indent=0]
----
-Since generators return a standard `Iterable`, you can consume them with a
-regular `for` loop or any Groovy collection method — no special syntax needed:
+The scope waits for every child to finish, even without explicit `await` calls:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=generator_regular_for,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=scope_waits,indent=0]
----
-With virtual threads (JDK 21+), blocking iteration is essentially free.
-On JDK 17-20, the generator runs on a platform thread from the cached pool.
+[[async-timeouts]]
+=== Timeouts and Delays
-[[async-for-await]]
-== Iterating with `for await`
+Apply a deadline to any task. If it doesn't complete in time, a
+`TimeoutException` is thrown:
-The `for await` loop works with any iterable source:
+[source,groovy]
+----
+include::../test/AsyncAwaitSpecTest.groovy[tags=timeout,indent=0]
+----
+
+Or use a fallback value instead of throwing:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=for_await_no_generator,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=timeout_fallback,indent=0]
----
-It also works with generators:
+`Awaitable.delay()` pauses without blocking a thread:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=for_await,indent=0]
+include::../test/AsyncAwaitSpecTest.groovy[tags=delay,indent=0]
----
-The key value of `for await` is with **reactive types** (Reactor `Flux`, RxJava
-`Observable`) where it automatically converts the source to a blocking iterable
-via the `AwaitableAdapter` SPI. Without `for await`, you would need to call the
-conversion manually (e.g., `flux.toIterable()` or `observable.blockingIterable()`).
-For generators and plain collections, a regular `for` loop works identically.
+[[async-adapters]]
+=== Framework Adapters
-[[async-defer]]
-== Deferred Cleanup with `defer`
+`await` natively understands `Awaitable`, `CompletableFuture`, `CompletionStage`,
+and `Future`. Third-party async types can be supported by implementing the
+`AwaitableAdapter` SPI and registering it via
+`META-INF/services/groovy.concurrent.AwaitableAdapter`.
-The `defer` keyword schedules a cleanup action to run when the enclosing
-`async` closure completes, regardless of whether it succeeds or throws.
-Deferred actions execute in LIFO (last-in, first-out) order, making it
-natural to pair resource acquisition with cleanup:
+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:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=defer_basic,indent=0]
+// Without groovy-rxjava — manual conversion
+def result = Single.just('hello').toCompletionStage().toCompletableFuture().join()
----
-Deferred actions always run, even when an exception occurs:
+With `groovy-rxjava` on the classpath, the conversion is transparent:
[source,groovy]
----
-include::../test/AsyncAwaitSpecTest.groovy[tags=defer_exception,indent=0]
+// With groovy-rxjava — adapter handles the plumbing
+def result = await Awaitable.from(Single.just('hello'))
----
-If a deferred action itself returns an `Awaitable` or `Future`, the result
-is awaited before the next deferred action runs, ensuring orderly cleanup
-of asynchronous resources.
-
[[async-executor]]
-== Executor Configuration
+=== Executor Configuration
By default, `async` uses:
-* **JDK 21+**: a virtual-thread-per-task executor (essentially unlimited,
- near-zero overhead per task)
-* **JDK 17–20**: a cached daemon thread pool (max 256 threads by default,
- configurable via the `groovy.async.parallelism` system property)
+* **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 for specific needs:
+You can override the executor:
[source,groovy]
----
import org.apache.groovy.runtime.async.AsyncSupport
import java.util.concurrent.Executors
-// Use a custom executor
AsyncSupport.setExecutor(Executors.newFixedThreadPool(4))
-
-// Reset to the default
-AsyncSupport.resetExecutor()
+AsyncSupport.resetExecutor() // restore default
----
-[[async-adapter]]
-== Framework Extensibility
+[[async-best-practices]]
+== Best Practices
-`await` natively understands `Awaitable`, `CompletableFuture`, `CompletionStage`,
-and `Future`. 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`
+[[async-prefer-values]]
+=== Prefer returning values over mutating shared state
-For example, without the adapter you must manually convert RxJava types:
+Async closures run on separate threads. Mutating shared variables from multiple
+closures is a race condition:
[source,groovy]
----
-// Without groovy-rxjava — manual conversion boilerplate
-def result = Single.just('hello')
- .toCompletionStage()
- .toCompletableFuture()
- .join()
-
-def items = []
-for (item in Observable.just(1, 2, 3).blockingIterable()) {
- items << item
-}
+// UNSAFE — shared mutation without synchronization
+var count = 0
+def tasks = (1..100).collect { async { count++ } }
+tasks.each { await it }
+// count may not be 100!
----
-With `groovy-rxjava` on the classpath, the adapter is auto-discovered and
-the conversion happens transparently:
+Instead, return values and collect results:
[source,groovy]
----
-// With groovy-rxjava — await and for-await handle the plumbing
-def result = await Awaitable.from(Single.just('hello'))
-
-def items = []
-for await (item in Observable.just(1, 2, 3)) {
- items << item
-}
+// 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.
+
+[[async-choose-right-tool]]
+=== Choosing the right tool
+
+[cols="2,3", options="header"]
+|===
+| 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.
+|===
+
[[async-summary]]
-== Summary of Keywords
+== Quick Reference
[cols="1,3"]
|===
-| Keyword | Description
-
-| `async { ... }` | Execute a closure asynchronously, returning an `Awaitable`
-| `await expr` | Block until the result is available; rethrows original exceptions
-| `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
-| `for await (x in source)` | Iterate over an async source (generator, Flux, Observable, etc.)
-| `defer expr` | Schedule a cleanup action (LIFO order) inside an `async` closure
-| `AsyncScope.withScope { scope -> ... }` | Structured concurrency scope — all children complete on exit
+| 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 three keywords (`async`, `await`, `defer`) are contextual — they can
-still be used as variable or method names in existing code.
+All keywords (`async`, `await`, `defer`) are contextual — they can still be
+used as variable or method names in existing code.
diff --git a/src/spec/test/AsyncAwaitSpecTest.groovy b/src/spec/test/AsyncAwaitSpecTest.groovy
index 3cfb7d10b5..6d4f3c2aaf 100644
--- a/src/spec/test/AsyncAwaitSpecTest.groovy
+++ b/src/spec/test/AsyncAwaitSpecTest.groovy
@@ -23,52 +23,79 @@ import static groovy.test.GroovyAssert.assertScript
class AsyncAwaitSpecTest {
+ // === Getting started ===
+
@Test
void testBasicAsyncAwait() {
assertScript '''
// tag::basic_async_await[]
def task = async { 21 * 2 }
- assert await(task()) == 42
+ assert await(task) == 42
// end::basic_async_await[]
'''
}
@Test
- void testSequentialWorkflow() {
+ void testDrawCard() {
assertScript '''
- // tag::sequential_workflow[]
- def fetchUserId = { String token -> 'user-42' }
- def fetchUserName = { String id -> 'Alice' }
- def loadProfile = { String name -> [name: name, level: 10] }
+ // tag::draw_card[]
+ def deck = ['2♠', '3♥', 'K♦', 'A♣']
+ def card = async { deck.shuffled()[0] }
+ println "You drew: ${await card}"
+ // end::draw_card[]
+ '''
+ }
- def task = async {
- var userId = fetchUserId('token-abc')
- var name = fetchUserName(userId)
- var profile = loadProfile(name)
- profile
+ @Test
+ void testExceptionHandling() {
+ assertScript '''
+ // tag::exception_handling[]
+ def drawFromEmpty = async {
+ throw new IllegalStateException('deck is empty')
}
- def profile = await task()
- assert profile.name == 'Alice'
- assert profile.level == 10
- // end::sequential_workflow[]
+ try {
+ await drawFromEmpty
+ } catch (IllegalStateException e) {
+ // Original exception — no CompletionException wrapper
+ assert e.message == 'deck is empty'
+ }
+ // end::exception_handling[]
+ '''
+ }
+
+ @Test
+ void testCfInterop() {
+ assertScript '''
+ import java.util.concurrent.CompletableFuture
+
+ // tag::cf_interop[]
+ // await works with CompletableFuture from Java libraries
+ def future = CompletableFuture.supplyAsync { 'A♠' }
+ assert await(future) == 'A♠'
+ // end::cf_interop[]
'''
}
+ // === Parallel tasks and combinators ===
+
@Test
- void testParallelTasks() {
+ void testDealHands() {
assertScript '''
import groovy.concurrent.Awaitable
- // tag::parallel_tasks[]
- def stats = async { [hp: 100, mp: 50] }
- def inventory = async { ['sword', 'shield'] }
- def villain = async { [name: 'Dragon', level: 20] }
+ // tag::deal_hands[]
+ // Deal cards to three players concurrently
+ def deck = ('A'..'K').collectMany { r -> ['♠','♥','♦','♣'].collect { "$r$it" } }.shuffled()
+ int i = 0
+ def draw5 = { -> deck[i..<(i+5)].tap { i += 5 } }
+
+ def alice = async { draw5() }
+ def bob = async { draw5() }
+ def carol = async { draw5() }
- def (s, inv, v) = await Awaitable.all(stats(), inventory(), villain())
- assert s.hp == 100
- assert inv.size() == 2
- assert v.name == 'Dragon'
- // end::parallel_tasks[]
+ def (a, b, c) = await Awaitable.all(alice, bob, carol)
+ assert a.size() == 5 && b.size() == 5 && c.size() == 5
+ // end::deal_hands[]
'''
}
@@ -80,25 +107,26 @@ class AsyncAwaitSpecTest {
def b = async { 2 }
def c = async { 3 }
- // These three forms are equivalent:
- def r1 = await(a(), b(), c()) // parenthesized multi-arg
- assert r1 == [1, 2, 3]
+ // Parenthesized multi-arg await (sugar for Awaitable.all):
+ def results = await(a, b, c)
+ assert results == [1, 2, 3]
// end::multi_arg_await[]
'''
}
@Test
- void testRacingTasks() {
+ void testFastestServer() {
assertScript '''
import groovy.concurrent.Awaitable
- // tag::racing_tasks[]
- def fast = async { 'fast' }
- def slow = async { Thread.sleep(500); 'slow' }
+ // tag::fastest_server[]
+ // Race two servers — use whichever responds first
+ def primary = async { Thread.sleep(200); 'primary-response' }
+ def fallback = async { 'fallback-response' }
- def winner = await Awaitable.any(fast(), slow())
- assert winner == 'fast'
- // end::racing_tasks[]
+ def response = await Awaitable.any(primary, fallback)
+ assert response == 'fallback-response'
+ // end::fastest_server[]
'''
}
@@ -108,12 +136,12 @@ class AsyncAwaitSpecTest {
import groovy.concurrent.Awaitable
// tag::first_success[]
- def failing = async { throw new RuntimeException('fail') }
- def succeeding = async { 'ok' }
+ // Try multiple sources — use the first that succeeds
+ def failing = async { throw new RuntimeException('server down') }
+ def succeeding = async { 'card-data-from-cache' }
- // Awaitable.first returns the first *successful* result
- def result = await Awaitable.first(failing(), succeeding())
- assert result == 'ok'
+ def result = await Awaitable.first(failing, succeeding)
+ assert result == 'card-data-from-cache'
// end::first_success[]
'''
}
@@ -124,91 +152,130 @@ class AsyncAwaitSpecTest {
import groovy.concurrent.Awaitable
// tag::all_settled[]
- def ok = async { 42 }
- def fail = async { throw new RuntimeException('boom') }
+ def save1 = async { 42 }
+ def save2 = async { throw new RuntimeException('db error') }
- def results = await Awaitable.allSettled(ok(), fail())
+ def results = await Awaitable.allSettled(save1, save2)
assert results[0].success && results[0].value == 42
- assert !results[1].success && results[1].error.message == 'boom'
+ assert !results[1].success && results[1].error.message == 'db error'
// end::all_settled[]
'''
}
+ // === Generators and for await ===
+
@Test
- void testExceptionHandling() {
+ void testDeckGenerator() {
assertScript '''
- // tag::exception_handling[]
- def task = async {
- throw new IOException('network error')
- }
- try {
- await task()
- } catch (IOException e) {
- // Original exception is rethrown — no wrapper layers
- assert e.message == 'network error'
+ // tag::deck_generator[]
+ 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"
+ }
+ }
}
- // end::exception_handling[]
+ def cards = dealCards.collect()
+ assert cards.size() == 52
+ assert cards.first() == 'A♠'
+ assert cards.last() == 'K♣'
+ // end::deck_generator[]
'''
}
@Test
- void testTimeout() {
+ void testForAwaitGenerator() {
assertScript '''
- import groovy.concurrent.Awaitable
- import java.util.concurrent.TimeoutException
-
- // tag::timeout[]
- def slow = async { Thread.sleep(5000); 'done' }
- try {
- await Awaitable.orTimeoutMillis(slow(), 100)
- } catch (TimeoutException e) {
- assert true // timed out as expected
+ // tag::for_await_generator[]
+ def topCards = async {
+ for (card in ['A♠', 'K♥', 'Q♦']) {
+ yield return card
+ }
}
- // end::timeout[]
+ def hand = []
+ for await (card in topCards) {
+ hand << card
+ }
+ assert hand == ['A♠', 'K♥', 'Q♦']
+ // end::for_await_generator[]
'''
}
@Test
- void testTimeoutWithFallback() {
+ void testForAwaitPlainCollection() {
assertScript '''
- import groovy.concurrent.Awaitable
+ // tag::for_await_collection[]
+ def results = []
+ for await (card in ['A♠', 'K♥', 'Q♦']) {
+ results << card
+ }
+ assert results == ['A♠', 'K♥', 'Q♦']
+ // end::for_await_collection[]
+ '''
+ }
- // tag::timeout_fallback[]
- def slow = async { Thread.sleep(5000); 'done' }
- def result = await Awaitable.completeOnTimeoutMillis(slow(), 'fallback', 100)
- assert result == 'fallback'
- // end::timeout_fallback[]
+ @Test
+ void testGeneratorRegularFor() {
+ assertScript '''
+ // tag::generator_regular_for[]
+ 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]
+ // end::generator_regular_for[]
'''
}
+ // === Channels ===
+
@Test
- void testDelay() {
+ void testChannel() {
assertScript '''
- import groovy.concurrent.Awaitable
+ import groovy.concurrent.AsyncChannel
- // tag::delay[]
- long start = System.currentTimeMillis()
- await Awaitable.delay(100)
- long elapsed = System.currentTimeMillis() - start
- assert elapsed >= 90
- // end::delay[]
+ // tag::channel[]
+ 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♣']
+ // end::channel[]
'''
}
+ // === Defer ===
+
@Test
void testDeferBasic() {
assertScript '''
// tag::defer_basic[]
def log = []
def task = async {
- defer { log << 'cleanup 1' }
- defer { log << 'cleanup 2' }
- log << 'work'
- 'done'
+ log << 'open connection'
+ defer { log << 'close connection' }
+ log << 'open transaction'
+ defer { log << 'close transaction' }
+ log << 'save game state'
+ 'saved'
}
- assert await(task()) == 'done'
- // Deferred actions run in LIFO order
- assert log == ['work', 'cleanup 2', 'cleanup 1']
+ 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']
// end::defer_basic[]
'''
}
@@ -220,58 +287,37 @@ class AsyncAwaitSpecTest {
def cleaned = false
def task = async {
defer { cleaned = true }
- throw new RuntimeException('oops')
+ throw new RuntimeException('save failed')
}
try {
- await task()
+ await task
} catch (RuntimeException e) {
- assert e.message == 'oops'
+ assert e.message == 'save failed'
}
- // Deferred actions still run even when an exception occurs
+ // Deferred actions run even when an exception occurs
assert cleaned
// end::defer_exception[]
'''
}
- @Test
- void testGoSpawn() {
- assertScript '''
- import groovy.concurrent.Awaitable
-
- // tag::go_spawn[]
- def task = Awaitable.go { 'spawned' }
- assert await(task) == 'spawned'
- // end::go_spawn[]
- '''
- }
+ // === Structured concurrency ===
@Test
- void testCompletableFutureInterop() {
- assertScript '''
- import java.util.concurrent.CompletableFuture
-
- // tag::cf_interop[]
- // await works with CompletableFuture from Java libraries
- def future = CompletableFuture.supplyAsync { 'from Java' }
- assert await(future) == 'from Java'
- // end::cf_interop[]
- '''
- }
-
- @Test
- void testStructuredConcurrency() {
+ void testTournamentScope() {
assertScript '''
import groovy.concurrent.AsyncScope
// tag::structured_concurrency[]
- def result = AsyncScope.withScope { scope ->
- def user = scope.async { [name: 'Alice'] }
- def orders = scope.async { ['order-1', 'order-2'] }
- [user: await(user), orders: await(orders)]
+ // 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)]
}
- // Both tasks guaranteed complete when withScope returns
- assert result.user.name == 'Alice'
- assert result.orders.size() == 2
+ // All tables guaranteed complete when withScope returns
+ assert results.size() == 3
+ assert results.max { it.score }.winner == 'Carol'
// end::structured_concurrency[]
'''
}
@@ -284,12 +330,12 @@ class AsyncAwaitSpecTest {
// tag::scope_fail_fast[]
try {
AsyncScope.withScope { scope ->
- scope.async { Thread.sleep(5000); 'slow' }
- scope.async { throw new RuntimeException('fail!') }
+ scope.async { Thread.sleep(5000); 'still playing' }
+ scope.async { throw new RuntimeException('player disconnected') }
}
} catch (RuntimeException e) {
- // The first failure cancels siblings and propagates
- assert e.message == 'fail!'
+ // First failure cancels all siblings and propagates
+ assert e.message == 'player disconnected'
}
// end::scope_fail_fast[]
'''
@@ -302,85 +348,59 @@ class AsyncAwaitSpecTest {
import java.util.concurrent.atomic.AtomicInteger
// tag::scope_waits[]
- def counter = new AtomicInteger(0)
+ def completed = new AtomicInteger(0)
AsyncScope.withScope { scope ->
- 3.times { scope.async { Thread.sleep(50); counter.incrementAndGet() } }
+ 3.times { scope.async { Thread.sleep(50); completed.incrementAndGet() } }
}
- // All children have completed
- assert counter.get() == 3
+ // All children have completed — even without explicit await
+ assert completed.get() == 3
// end::scope_waits[]
'''
}
- @Test
- void testYieldReturn() {
- assertScript '''
- // tag::yield_return[]
- def fibonacci = async {
- long a = 0, b = 1
- for (i in 1..10) {
- yield return a
- (a, b) = [b, a + b]
- }
- }
- assert fibonacci().collect() == [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
- // end::yield_return[]
- '''
- }
+ // === Timeouts and delays ===
@Test
- void testGeneratorWithRegularFor() {
+ void testTimeout() {
assertScript '''
- // tag::generator_regular_for[]
- def countdown = async {
- for (i in 5..1) {
- yield return i
- }
- }
- // Generators return Iterable — regular for loop works
- def results = []
- for (n in countdown()) {
- results << n
- }
- assert results == [5, 4, 3, 2, 1]
+ import groovy.concurrent.Awaitable
+ import java.util.concurrent.TimeoutException
- // Collection methods work too
- assert countdown().collect { it * 10 } == [50, 40, 30, 20, 10]
- // end::generator_regular_for[]
+ // tag::timeout[]
+ 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
+ }
+ // end::timeout[]
'''
}
@Test
- void testForAwaitWithoutGenerator() {
+ void testTimeoutFallback() {
assertScript '''
- import java.util.concurrent.CompletableFuture
+ import groovy.concurrent.Awaitable
- // tag::for_await_no_generator[]
- // for await works with any iterable — no generator needed
- def results = []
- for await (item in ['alpha', 'beta', 'gamma']) {
- results << item.toUpperCase()
- }
- assert results == ['ALPHA', 'BETA', 'GAMMA']
- // end::for_await_no_generator[]
+ // tag::timeout_fallback[]
+ def slowPlayer = async { Thread.sleep(5000); 'deliberate move' }
+ def move = await Awaitable.completeOnTimeoutMillis(slowPlayer, 'auto-pass', 100)
+ assert move == 'auto-pass'
+ // end::timeout_fallback[]
'''
}
@Test
- void testForAwait() {
+ void testDelay() {
assertScript '''
- // tag::for_await[]
- def squares = async {
- for (i in 1..5) {
- yield return i * i
- }
- }
- def results = []
- for await (n in squares()) {
- results << n
- }
- assert results == [1, 4, 9, 16, 25]
- // end::for_await[]
+ import groovy.concurrent.Awaitable
+
+ // tag::delay[]
+ long start = System.currentTimeMillis()
+ await Awaitable.delay(100) // pause without blocking a thread
+ assert System.currentTimeMillis() - start >= 90
+ // end::delay[]
'''
}
}
diff --git a/src/test/groovy/groovy/AsyncAwaitTest.groovy b/src/test/groovy/groovy/AsyncAwaitTest.groovy
index a96636daa4..ca2d53894f 100644
--- a/src/test/groovy/groovy/AsyncAwaitTest.groovy
+++ b/src/test/groovy/groovy/AsyncAwaitTest.groovy
@@ -28,12 +28,10 @@ final class AsyncAwaitTest {
// === Layer 1: basic async/await ===
@Test
- void testAsyncClosureAndAwait() {
+ void testAsyncAndAwait() {
assertScript '''
- import groovy.concurrent.Awaitable
-
def task = async { 21 * 2 }
- def result = await task()
+ def result = await task
assert result == 42
'''
}
@@ -54,7 +52,7 @@ final class AsyncAwaitTest {
assertScript '''
def task = async { throw new IOException('test error') }
try {
- await task()
+ await task
} catch (IOException e) {
assert e.message == 'test error'
return
@@ -71,23 +69,14 @@ final class AsyncAwaitTest {
def counter = new AtomicInteger(0)
def task1 = async { Thread.sleep(50); counter.incrementAndGet(); 'a' }
def task2 = async { Thread.sleep(50); counter.incrementAndGet(); 'b' }
- def r1 = await task1()
- def r2 = await task2()
+ def r1 = await task1
+ def r2 = await task2
assert counter.get() == 2
assert r1 == 'a'
assert r2 == 'b'
'''
}
- @Test
- void testAsyncWithClosureArgs() {
- assertScript '''
- def multiply = async { x, y -> x * y }
- def result = await multiply(6, 7)
- assert result == 42
- '''
- }
-
@Test
void testAwaitNull() {
assertScript '''
@@ -98,7 +87,6 @@ final class AsyncAwaitTest {
@Test
void testAsyncKeywordAsVariable() {
- // 'async' can still be used as a variable name
assertScript '''
def async = 'hello'
assert async.toUpperCase() == 'HELLO'
@@ -121,7 +109,7 @@ final class AsyncAwaitTest {
def a = async { 1 }
def b = async { 2 }
def c = async { 3 }
- def results = await(a(), b(), c())
+ def results = await(a, b, c)
assert results == [1, 2, 3]
'''
}
@@ -133,7 +121,7 @@ final class AsyncAwaitTest {
def a = async { 'x' }
def b = async { 'y' }
- def results = await Awaitable.all(a(), b())
+ def results = await Awaitable.all(a, b)
assert results == ['x', 'y']
'''
}
@@ -145,7 +133,7 @@ final class AsyncAwaitTest {
def fast = async { 'fast' }
def slow = async { Thread.sleep(500); 'slow' }
- def result = await Awaitable.any(fast(), slow())
+ def result = await Awaitable.any(fast, slow)
assert result == 'fast'
'''
}
@@ -157,7 +145,7 @@ final class AsyncAwaitTest {
def fail1 = async { throw new RuntimeException('fail') }
def success = async { 'ok' }
- def result = await Awaitable.first(fail1(), success())
+ def result = await Awaitable.first(fail1, success)
assert result == 'ok'
'''
}
@@ -169,7 +157,7 @@ final class AsyncAwaitTest {
def ok = async { 42 }
def fail = async { throw new RuntimeException('boom') }
- def results = await Awaitable.allSettled(ok(), fail())
+ def results = await Awaitable.allSettled(ok, fail)
assert results.size() == 2
assert results[0].success
assert results[0].value == 42
@@ -186,7 +174,7 @@ final class AsyncAwaitTest {
long start = System.currentTimeMillis()
await Awaitable.delay(100)
long elapsed = System.currentTimeMillis() - start
- assert elapsed >= 90 // allow small timing variance
+ assert elapsed >= 90
'''
}
@@ -198,7 +186,7 @@ final class AsyncAwaitTest {
def slow = async { Thread.sleep(5000); 'done' }
try {
- await Awaitable.orTimeoutMillis(slow(), 100)
+ await Awaitable.orTimeoutMillis(slow, 100)
assert false : 'Should have timed out'
} catch (TimeoutException e) {
assert true
@@ -212,7 +200,7 @@ final class AsyncAwaitTest {
import groovy.concurrent.Awaitable
def slow = async { Thread.sleep(5000); 'done' }
- def result = await Awaitable.completeOnTimeoutMillis(slow(), 'fallback', 100)
+ def result = await Awaitable.completeOnTimeoutMillis(slow, 'fallback', 100)
assert result == 'fallback'
'''
}
@@ -261,7 +249,6 @@ final class AsyncAwaitTest {
void testScopeFailFastCancelsSiblings() {
assertScript '''
import groovy.concurrent.AsyncScope
- import java.util.concurrent.CancellationException
def slowStarted = new java.util.concurrent.CountDownLatch(1)
def result = null
@@ -297,7 +284,6 @@ final class AsyncAwaitTest {
scope.async { Thread.sleep(50); counter.incrementAndGet() }
scope.async { Thread.sleep(50); counter.incrementAndGet() }
}
- // All three must have completed by the time withScope returns
assert counter.get() == 3
'''
}
@@ -325,8 +311,6 @@ final class AsyncAwaitTest {
assert AsyncScope.current() == null
AsyncScope.withScope { scope ->
- // Inside the scope, current() returns the scope
- // (but we're on the calling thread, not a child thread)
def childSaw = scope.async {
AsyncScope.current() != null
}
@@ -373,13 +357,13 @@ final class AsyncAwaitTest {
@Test
void testYieldReturnBasic() {
assertScript '''
- def gen = async {
+ def items = async {
yield return 1
yield return 2
yield return 3
}
def results = []
- for (item in gen()) {
+ for (item in items) {
results << item
}
assert results == [1, 2, 3]
@@ -389,37 +373,121 @@ final class AsyncAwaitTest {
@Test
void testYieldReturnWithLoop() {
assertScript '''
- def range = async {
+ def items = async {
for (i in 1..5) {
yield return i * 10
}
}
- assert range().collect() == [10, 20, 30, 40, 50]
+ assert items.collect() == [10, 20, 30, 40, 50]
'''
}
+ // === channels ===
+
@Test
- void testYieldReturnWithArgs() {
+ void testChannelBasic() {
assertScript '''
- def repeat = async { value, times ->
- for (i in 1..times) {
- yield return value
+ import groovy.concurrent.AsyncChannel
+
+ def ch = AsyncChannel.create(2)
+ async {
+ await ch.send('a')
+ await ch.send('b')
+ ch.close()
+ }
+ def results = []
+ for await (item in ch) {
+ results << item
+ }
+ assert results == ['a', 'b']
+ '''
+ }
+
+ @Test
+ void testChannelUnbuffered() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+
+ def ch = AsyncChannel.create() // rendezvous
+ async {
+ await ch.send(1)
+ await ch.send(2)
+ await ch.send(3)
+ ch.close()
+ }
+ def results = []
+ for (item in ch) {
+ results << item
+ }
+ assert results == [1, 2, 3]
+ '''
+ }
+
+ @Test
+ void testChannelProducerConsumer() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+
+ def ch = AsyncChannel.create(3)
+ // Producer
+ async {
+ for (i in 1..5) {
+ await ch.send(i * 10)
}
+ ch.close()
+ }
+ // Consumer
+ def sum = 0
+ for await (value in ch) {
+ sum += value
+ }
+ assert sum == 150
+ '''
+ }
+
+ @Test
+ void testChannelClosedForSend() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+ import groovy.concurrent.ChannelClosedException
+
+ def ch = AsyncChannel.create(1)
+ ch.close()
+ try {
+ await ch.send('too late')
+ assert false : 'Should have thrown'
+ } catch (ChannelClosedException e) {
+ assert e.message.contains('closed')
}
- assert repeat('x', 3).collect() == ['x', 'x', 'x']
+ '''
+ }
+
+ @Test
+ void testChannelQueryMethods() {
+ assertScript '''
+ import groovy.concurrent.AsyncChannel
+
+ def ch = AsyncChannel.create(5)
+ assert ch.capacity == 5
+ assert ch.bufferedSize == 0
+ assert !ch.closed
+ await ch.send('x')
+ assert ch.bufferedSize == 1
+ ch.close()
+ assert ch.closed
'''
}
@Test
void testYieldReturnExceptionPropagation() {
assertScript '''
- def gen = async {
+ def items = async {
yield return 1
throw new RuntimeException('generator error')
}
def results = []
try {
- for (item in gen()) {
+ for (item in items) {
results << item
}
assert false : 'Should have thrown'
@@ -446,13 +514,13 @@ final class AsyncAwaitTest {
@Test
void testForAwaitWithGenerator() {
assertScript '''
- def gen = async {
+ def items = async {
yield return 'a'
yield return 'b'
yield return 'c'
}
def results = []
- for await (item in gen()) {
+ for await (item in items) {
results << item.toUpperCase()
}
assert results == ['A', 'B', 'C']
@@ -471,19 +539,18 @@ final class AsyncAwaitTest {
'''
}
- // === regular for loop with generators (no for-await needed) ===
+ // === regular for loop with generators ===
@Test
void testRegularForLoopWithGenerator() {
- // Generators return Iterable, so regular for loop works too
assertScript '''
- def gen = async {
+ def items = async {
yield return 'a'
yield return 'b'
yield return 'c'
}
def results = []
- for (item in gen()) {
+ for (item in items) {
results << item
}
assert results == ['a', 'b', 'c']
@@ -492,15 +559,13 @@ final class AsyncAwaitTest {
@Test
void testCollectWithGenerator() {
- // Standard Groovy collection methods work on generator output
assertScript '''
def squares = async {
for (i in 1..5) {
yield return i * i
}
}
- assert squares().collect() == [1, 4, 9, 16, 25]
- assert squares().collect { it * 2 } == [2, 8, 18, 32, 50]
+ assert squares.collect() == [1, 4, 9, 16, 25]
'''
}
@@ -515,7 +580,7 @@ final class AsyncAwaitTest {
log << 'work'
'result'
}
- def result = await task()
+ def result = await task
assert result == 'result'
assert log == ['work', 'cleanup']
'''
@@ -531,7 +596,7 @@ final class AsyncAwaitTest {
log << 'body'
'done'
}
- await task()
+ await task
assert log == ['body', 'second registered, first to run', 'first registered, last to run']
'''
}
@@ -545,7 +610,7 @@ final class AsyncAwaitTest {
throw new RuntimeException('oops')
}
try {
- await task()
+ await task
} catch (RuntimeException e) {
assert e.message == 'oops'
}
@@ -562,7 +627,7 @@ final class AsyncAwaitTest {
log << 'body'
42
}
- assert await(task()) == 42
+ assert await(task) == 42
assert log == ['body', 'deferred closure']
'''
}
@@ -575,7 +640,6 @@ final class AsyncAwaitTest {
import org.apache.groovy.runtime.async.AsyncSupport
import java.util.concurrent.Executors
- def threadNames = Collections.synchronizedList([])
def exec = Executors.newFixedThreadPool(2) { r ->
def t = new Thread(r, 'custom-pool')
t.daemon = true
@@ -584,7 +648,7 @@ final class AsyncAwaitTest {
try {
AsyncSupport.setExecutor(exec)
def task = async { Thread.currentThread().name }
- def name = await task()
+ def name = await task
assert name == 'custom-pool'
} finally {
AsyncSupport.resetExecutor()
@@ -598,7 +662,6 @@ final class AsyncAwaitTest {
assertScript '''
import org.apache.groovy.runtime.async.AsyncSupport
- // Just verify this doesn't throw — actual value depends on JDK version
def available = AsyncSupport.isVirtualThreadsAvailable()
assert available instanceof Boolean
'''