Minor tweaks

Author: daniellansun

src/main/java/groovy/concurrent/AsyncStream.java

@@ -75,6 +75,36 @@ public interface AsyncStream<T> extends AutoCloseable {
     default void close() {
     }
 
+    /**
+     * Converts the given source to an {@code AsyncStream}.
+     * <p>
+     * If the source is already an {@code AsyncStream}, it is returned as-is.
+     * Otherwise, the {@link AwaitableAdapterRegistry} is consulted to find a
+     * suitable adapter. Built-in adapters handle {@link Iterable} and
+     * {@link java.util.Iterator}; the auto-discovered {@code FlowPublisherAdapter}
+     * handles {@link java.util.concurrent.Flow.Publisher}; third-party frameworks
+     * can register additional adapters via the registry.
+     * <p>
+     * This is the recommended entry point for converting external collection or
+     * reactive types to {@code AsyncStream}:
+     * <pre>
+     * AsyncStream&lt;String&gt; stream = AsyncStream.from(myList)
+     * AsyncStream&lt;Integer&gt; stream2 = AsyncStream.from(myFlowPublisher)
+     * </pre>
+     *
+     * @param source the source object; must not be {@code null}
+     * @param <T>    the element type
+     * @return an async stream backed by the source
+     * @throws IllegalArgumentException if {@code source} is {@code null}
+     *         or no adapter supports the source type
+     * @see AwaitableAdapterRegistry#toAsyncStream(Object)
+     * @since 6.0.0
+     */
+    @SuppressWarnings("unchecked")
+    static <T> AsyncStream<T> from(Object source) {
+        return AwaitableAdapterRegistry.toAsyncStream(source);
+    }
+
     /**
      * Returns an empty {@code AsyncStream} that completes immediately.
      */

src/main/java/groovy/concurrent/Awaitable.java

@@ -54,14 +54,18 @@
  *       {@code Promise.allSettled()}</li>
  *   <li>{@link #delay(long) Awaitable.delay(ms)} — like
  *       {@code Task.Delay()} / {@code setTimeout}</li>
- *   <li>{@link #timeout(Object, long) Awaitable.timeout(task, ms)} — like
+ *   <li>{@link #orTimeoutMillis(Object, long) Awaitable.orTimeoutMillis(task, ms)} — like
  *       Kotlin's {@code withTimeout} or a JavaScript promise raced against a timer</li>
- *   <li>{@link #timeoutOr(Object, Object, long) Awaitable.timeoutOr(task, fallback, ms)} —
+ *   <li>{@link #completeOnTimeoutMillis(Object, Object, long)
+ *       Awaitable.completeOnTimeoutMillis(task, fallback, ms)} —
  *       like a timeout with fallback/default value</li>
  * </ul>
  * <p>
- * <b>Static factories:</b>
+ * <b>Static factories and conversion:</b>
  * <ul>
+ *   <li>{@link #from(Object) Awaitable.from(source)} — converts any supported
+ *       async type (CompletableFuture, CompletionStage, Future, Flow.Publisher, etc.)
+ *       to an {@code Awaitable}</li>
  *   <li>{@link #of(Object) Awaitable.of(value)} — like
  *       {@code Task.FromResult()} / {@code Promise.resolve()}</li>
  *   <li>{@link #failed(Throwable) Awaitable.failed(error)} — like
@@ -297,6 +301,35 @@ default Awaitable<T> completeOnTimeout(T fallback, long duration, TimeUnit unit)
 
     // ---- Static factories ----
 
+    /**
+     * Converts the given source to an {@code Awaitable}.
+     * <p>
+     * If the source is already an {@code Awaitable}, it is returned as-is.
+     * Otherwise, the {@link AwaitableAdapterRegistry} is consulted to find a
+     * suitable adapter. Built-in adapters handle {@link CompletableFuture},
+     * {@link java.util.concurrent.CompletionStage}, {@link java.util.concurrent.Future},
+     * and {@link java.util.concurrent.Flow.Publisher}; third-party frameworks
+     * can register additional adapters via the registry.
+     * <p>
+     * This is the recommended entry point for converting external async types
+     * to {@code Awaitable}:
+     * <pre>
+     * Awaitable&lt;String&gt; aw = Awaitable.from(someCompletableFuture)
+     * Awaitable&lt;Integer&gt; aw2 = Awaitable.from(someReactorMono)
+     * </pre>
+     *
+     * @param source the source object; must not be {@code null}
+     * @param <T>    the result type
+     * @return an awaitable backed by the source
+     * @throws IllegalArgumentException if {@code source} is {@code null}
+     *         or no adapter supports the source type
+     * @see AwaitableAdapterRegistry#toAwaitable(Object)
+     * @since 6.0.0
+     */
+    static <T> Awaitable<T> from(Object source) {
+        return AwaitableAdapterRegistry.toAwaitable(source);
+    }
+
     /**
      * Returns an already-completed {@code Awaitable} with the given value.
      * Analogous to C#'s {@code Task.FromResult()} or JavaScript's

src/main/java/groovy/concurrent/AwaitableAdapterRegistry.java

@@ -111,13 +111,18 @@ public static void setBlockingExecutor(Executor executor) {
     /**
      * Converts the given source to an {@link Awaitable}.
      * If the source is already an {@code Awaitable}, it is returned as-is.
+     * <p>
+     * <b>Tip:</b> user code should generally prefer {@link Awaitable#from(Object)},
+     * which delegates to this method but is more discoverable from the
+     * {@code Awaitable} type itself.
      *
      * @param source the source object; must not be {@code null}
      * @throws IllegalArgumentException if {@code source} is {@code null}
      *         or no adapter supports the source type
+     * @see Awaitable#from(Object)
      */
     @SuppressWarnings("unchecked")
-    public static <T> Awaitable<T> toAwaitable(Object source) {
+    static <T> Awaitable<T> toAwaitable(Object source) {
         if (source == null) {
             throw new IllegalArgumentException("Cannot convert null to Awaitable");
         }
@@ -136,13 +141,18 @@ public static <T> Awaitable<T> toAwaitable(Object source) {
     /**
      * Converts the given source to an {@link AsyncStream}.
      * If the source is already an {@code AsyncStream}, it is returned as-is.
+     * <p>
+     * <b>Tip:</b> user code should generally prefer {@link AsyncStream#from(Object)},
+     * which delegates to this method but is more discoverable from the
+     * {@code AsyncStream} type itself.
      *
      * @param source the source object; must not be {@code null}
      * @throws IllegalArgumentException if {@code source} is {@code null}
      *         or no adapter supports the source type
+     * @see AsyncStream#from(Object)
      */
     @SuppressWarnings("unchecked")
-    public static <T> AsyncStream<T> toAsyncStream(Object source) {
+    static <T> AsyncStream<T> toAsyncStream(Object source) {
         if (source == null) {
             throw new IllegalArgumentException("Cannot convert null to AsyncStream");
         }

src/main/java/groovy/transform/Async.java

@@ -107,7 +107,9 @@
  * <ul>
  *   <li>Cannot be applied to abstract methods</li>
  *   <li>Cannot be applied to constructors</li>
- *   <li>Cannot be applied to methods that already return {@code Awaitable}</li>
+ *   <li>Cannot be applied to methods that already return an async type
+ *       ({@code Awaitable}, {@code AsyncStream}, {@code CompletableFuture},
+ *       {@code CompletionStage}, or {@code Future})</li>
  * </ul>
  *
  * @see groovy.concurrent.Awaitable

src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java

@@ -932,6 +932,10 @@ public ExpressionStatement visitYieldReturnStmtAlt(final YieldReturnStmtAltConte
 
     @Override
     public ExpressionStatement visitDeferStmtAlt(final DeferStmtAltContext ctx) {
+        if (asyncContextDepth == 0) {
+            throw createParsingFailedException(
+                    "`defer` can only be used inside an async method or async closure", ctx);
+        }
         Expression action;
         ExpressionStatement stmtExprStmt = (ExpressionStatement) this.visit(ctx.statementExpression());
         Expression expr = stmtExprStmt.getExpression();
@@ -1773,8 +1777,15 @@ public MethodNode visitMethodDeclaration(final MethodDeclarationContext ctx) {
         Parameter[] parameters = this.visitFormalParameters(ctx.formalParameters());
         ClassNode[] exceptions = this.visitQualifiedClassNameList(ctx.qualifiedClassNameList());
 
+        boolean isAsync = modifierManager.containsAny(ASYNC);
+        if (isAsync) asyncContextDepth++;
         anonymousInnerClassesDefinedInMethodStack.push(new LinkedList<>());
-        Statement code = this.visitMethodBody(ctx.methodBody());
+        Statement code;
+        try {
+            code = this.visitMethodBody(ctx.methodBody());
+        } finally {
+            if (isAsync) asyncContextDepth--;
+        }
         List<InnerClassNode> anonymousInnerClassList = anonymousInnerClassesDefinedInMethodStack.pop();
 
         MethodNode methodNode;
@@ -1789,7 +1800,7 @@ public MethodNode visitMethodDeclaration(final MethodDeclarationContext ctx) {
         }
 
         // Inject @Async annotation for methods declared with the 'async' keyword modifier
-        if (modifierManager.containsAny(ASYNC)) {
+        if (isAsync) {
             methodNode.addAnnotation(new AnnotationNode(ClassHelper.make("groovy.transform.Async")));
         }
 
@@ -2985,7 +2996,13 @@ public Expression visitAwaitExprAlt(final AwaitExprAltContext ctx) {
 
     @Override
     public Expression visitAsyncClosureExprAlt(final AsyncClosureExprAltContext ctx) {
-        ClosureExpression closure = this.visitClosureOrLambdaExpression(ctx.closureOrLambdaExpression());
+        asyncContextDepth++;
+        ClosureExpression closure;
+        try {
+            closure = this.visitClosureOrLambdaExpression(ctx.closureOrLambdaExpression());
+        } finally {
+            asyncContextDepth--;
+        }
         boolean hasUserParams = closure.getParameters() != null && closure.getParameters().length > 0;
         boolean hasYieldReturn = AsyncTransformHelper.containsYieldReturn(closure.getCode());
         boolean hasDefer = AsyncTransformHelper.containsDefer(closure.getCode());
@@ -4886,6 +4903,9 @@ public List<DeclarationExpression> getDeclarationExpressions() {
     private int visitingClosureCount;
     private int visitingAssertStatementCount;
     private int visitingArrayInitializerCount;
+    /** Tracks nesting depth of async contexts (async methods and async closures)
+     *  to validate that {@code defer} is only used within an async body. */
+    private int asyncContextDepth;
 
     private static final int SLL_THRESHOLD = SystemUtil.getIntegerSafe("groovy.antlr4.sll.threshold", -1);
 

src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java

@@ -21,7 +21,6 @@
 import groovy.concurrent.AsyncStream;
 import groovy.concurrent.AwaitResult;
 import groovy.concurrent.Awaitable;
-import groovy.concurrent.AwaitableAdapterRegistry;
 import groovy.lang.Closure;
 
 import java.lang.invoke.MethodHandle;
@@ -64,8 +63,9 @@
  * <ul>
  *   <li><b>Async execution</b> — {@code executeAsync()}, {@code executeAsyncVoid()},
  *       and {@code wrapAsync()} run closures on the configured executor</li>
- *   <li><b>Await</b> — all {@code await()} overloads go through the
- *       {@link AwaitableAdapterRegistry} so that third-party async types
+ *   <li><b>Await</b> — all {@code await()} overloads use
+ *       {@link groovy.concurrent.Awaitable#from(Object) Awaitable.from()} so
+ *       that third-party async types
  *       (RxJava {@code Single}, Reactor {@code Mono}, etc.) are supported
  *       transparently once an adapter is registered</li>
  *   <li><b>Async generators</b> — {@code generateAsyncStream()} manages the
@@ -112,7 +112,6 @@
  *
  * @see groovy.concurrent.Awaitable
  * @see groovy.transform.Async
- * @see Awaitable
  * @since 6.0.0
  */
 public class AsyncSupport {
@@ -267,8 +266,8 @@ public static <T> T await(Future<T> future) {
     }
 
     /**
-     * Awaits an arbitrary object by adapting it to {@link Awaitable} via the
-     * {@link AwaitableAdapterRegistry}.  This is the fallback overload called
+     * Awaits an arbitrary object by adapting it to {@link Awaitable} via
+     * {@link Awaitable#from(Object)}.  This is the fallback overload called
      * by the {@code await} expression when the compile-time type is not one
      * of the other supported types.  Returns {@code null} for a {@code null}
      * argument.
@@ -285,7 +284,7 @@ public static <T> T await(Object source) {
         if (source instanceof CompletionStage) return await((CompletionStage<T>) source);
         if (source instanceof Future) return await((Future<T>) source);
         if (source instanceof Closure) return awaitClosure((Closure<?>) source);
-        return await(AwaitableAdapterRegistry.<T>toAwaitable(source));
+        return await(Awaitable.<T>from(source));
     }
 
     /**
@@ -543,7 +542,7 @@ private static CompletableFuture<?> toCompletableFuture(Object source) {
         if (source instanceof CompletableFuture<?> cf) return cf;
         if (source instanceof Awaitable<?> a) return a.toCompletableFuture();
         if (source instanceof CompletionStage<?> cs) return cs.toCompletableFuture();
-        return AwaitableAdapterRegistry.toAwaitable(source).toCompletableFuture();
+        return Awaitable.from(source).toCompletableFuture();
     }
 
     // ---- non-blocking combinators (return Awaitable) --------------------
@@ -813,7 +812,7 @@ public static Awaitable<Void> delay(long duration, TimeUnit unit) {
     public static <T> AsyncStream<T> toAsyncStream(Object source) {
         if (source == null) return AsyncStream.empty();
         if (source instanceof AsyncStream) return (AsyncStream<T>) source;
-        return AwaitableAdapterRegistry.toAsyncStream(source);
+        return AsyncStream.from(source);
     }
 
     /**

src/main/java/org/apache/groovy/runtime/async/FlowPublisherAdapter.java

@@ -64,9 +64,13 @@
  * <ul>
  *   <li>§2.5 — duplicate {@code onSubscribe} cancels the second subscription</li>
  *   <li>§2.13 — {@code null} items in {@code onNext} are rejected immediately</li>
- *   <li>Terminal signals ({@code onError}/{@code onComplete}) use
+ *   <li>All signals ({@code onNext}, {@code onError}, {@code onComplete}) use
  *       blocking {@code put()} to guarantee delivery even under queue
  *       contention</li>
+ *   <li>Back-pressure is enforced by requesting exactly one item after
+ *       each consumed element; demand is signalled <em>before</em> the
+ *       consumer's {@code moveNext()} awaitable completes, preventing
+ *       livelock when producer and consumer share the same thread pool</li>
  * </ul>
  *
  * @see groovy.concurrent.AwaitableAdapterRegistry
@@ -209,14 +213,10 @@ public void onComplete() {
     // Signal wrapper types allow us to distinguish values, errors, and
     // completion in a single queue without type confusion.
 
-    private static final class ValueSignal<T> {
-        final T value;
-        ValueSignal(T value) { this.value = value; }
+    private record ValueSignal<T>(T value) {
     }
 
-    private static final class ErrorSignal {
-        final Throwable error;
-        ErrorSignal(Throwable error) { this.error = error; }
+    private record ErrorSignal(Throwable error) {
     }
 
     /** Singleton sentinel for stream completion. */
@@ -227,9 +227,16 @@ private static final class ErrorSignal {
      * providing a pull-based iteration interface over a push-based source.
      *
      * <p>Back-pressure is enforced by requesting exactly one item after
-     * each consumed element.  The internal bounded queue (capacity
-     * {@value QUEUE_CAPACITY}) absorbs minor timing jitter between
-     * producer and consumer.</p>
+     * each consumed element.  Demand is signalled <em>before</em> the
+     * consumer's {@code moveNext()} awaitable completes, so the publisher
+     * can begin producing the next value while the consumer processes the
+     * current one — this prevents livelock when producer and consumer
+     * share the same thread pool.</p>
+     *
+     * <p>The internal bounded queue (capacity {@value QUEUE_CAPACITY})
+     * absorbs minor timing jitter between producer and consumer.  All
+     * signals use blocking {@code put()}, ensuring no items or terminal
+     * events are silently dropped.</p>
      *
      * <p><b>Resource management:</b> When the consumer calls
      * {@link AsyncStream#close()} (e.g. via {@code break} in a
@@ -277,10 +284,16 @@ public void onNext(T item) {
                     return;
                 }
                 if (!closedRef.get()) {
-                    // Use offer() for value signals — if the queue is full (publisher
-                    // misbehaving with respect to demand), the item is dropped rather
-                    // than blocking the publisher thread indefinitely.
-                    queue.offer(new ValueSignal<>(item));
+                    try {
+                        // Blocking put() guarantees the item reaches the consumer.
+                        // Since demand is capped at 1 (one request(1) per moveNext),
+                        // a well-behaved publisher will never overflow the queue; put()
+                        // still protects against misbehaving publishers by blocking
+                        // rather than silently dropping the value.
+                        queue.put(new ValueSignal<>(item));
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                    }
                 }
             }
 
@@ -326,10 +339,16 @@ public Awaitable<Boolean> moveNext() {
 
                     if (signal instanceof ValueSignal) {
                         current = ((ValueSignal<T>) signal).value;
-                        cf.complete(Boolean.TRUE);
-                        // Request the next item — back-pressure: one-at-a-time
+                        // Signal demand for the next item BEFORE completing
+                        // the awaitable, so the publisher can begin producing
+                        // the next value while the consumer processes this one.
+                        // Ordering here is critical: if request(1) were called
+                        // after cf.complete(), the consumer could re-enter
+                        // moveNext() and block in take() before demand was
+                        // signalled, creating a livelock.
                         Flow.Subscription sub = subRef.get();
                         if (sub != null) sub.request(1);
+                        cf.complete(Boolean.TRUE);
                     } else if (signal instanceof ErrorSignal) {
                         streamClosed.set(true);
                         cf.completeExceptionally(((ErrorSignal) signal).error);
@@ -366,10 +385,14 @@ public void close() {
                     if (sub != null) sub.cancel();
                     // Drain the queue and inject a sentinel to unblock a
                     // concurrent moveNext() that may be blocked in take().
-                    // Using offer() after clear() performs a non-blocking,
-                    // best-effort delivery of the sentinel to any waiter.
+                    // Using blocking put() after clear() guarantees delivery;
+                    // since the queue is freshly cleared, put() will not block.
                     queue.clear();
-                    queue.offer(COMPLETE_SENTINEL);
+                    try {
+                        queue.put(COMPLETE_SENTINEL);
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                    }
                 }
             }
         };

src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java

@@ -147,7 +147,12 @@ public static Expression buildYieldReturnCall(Expression arg) {
 
     /**
      * Builds {@code AsyncSupport.defer($__deferScope__, action)}.
-     * The synthetic defer-scope variable is injected automatically.
+     * The synthetic defer-scope variable is injected by
+     * {@link #wrapWithDeferScope(Statement)} during AST transformation.
+     * <p>
+     * The parser validates that {@code defer} only appears inside an async
+     * context (async method or async closure); using it elsewhere is a
+     * compile-time error.
      *
      * @param action the deferred action expression (typically a closure)
      * @return an AST node representing the static call