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/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
@@ -227,9 +231,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 +288,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 +343,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 +389,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/spec/doc/core-async-await.adoc

@@ -610,7 +610,10 @@ RxJava 3 via adapters, Vert.x, etc.).
 
 The internal adapter uses a bounded buffer (capacity 256) to bridge the push-based `Publisher`
 model to Groovy's pull-based `AsyncStream` model, preventing unbounded memory growth with
-fast publishers while maintaining throughput.
+fast publishers while maintaining throughput.  Back-pressure is enforced by requesting exactly
+one item at a time; demand for the next element is signalled _before_ the consumer's
+`moveNext()` awaitable completes, so the publisher can begin producing the next value while
+the consumer processes the current one.
 
 === Awaiting a Single Value
 
@@ -638,6 +641,24 @@ This is conceptually similar to JavaScript async iterators' `return()`, C#'s
 [[adapter-registry]]
 == Adapter Registry
 
+=== Converting External Types with `from()`
+
+The `Awaitable.from(source)` and `AsyncStream.from(source)` static methods are the recommended
+entry points for converting external async types to Groovy's native abstractions. They delegate
+to the `AwaitableAdapterRegistry` internally but provide a more discoverable, user-friendly API:
+
+[source,groovy]
+----
+include::../test/AsyncAwaitSpecTest.groovy[tags=from_conversion,indent=0]
+----
+
+`Awaitable.from()` accepts any type supported by a registered adapter: `CompletableFuture`,
+`CompletionStage`, `Future`, `Flow.Publisher`, or any third-party type with a custom adapter.
+If the source is already an `Awaitable`, it is returned as-is. `AsyncStream.from()` works
+similarly for multi-value sources such as `Iterable`, `Iterator`, or `Flow.Publisher`.
+
+=== Built-in Adapters
+
 The `groovy.concurrent.AwaitableAdapterRegistry` allows extending `await` to support additional
 asynchronous types from third-party frameworks. The built-in adapter handles:
 
@@ -959,7 +980,9 @@ writes (adapter registration) are rare, reads (every `await`) are frequent
 * `AsyncStreamGenerator` close state — `AtomicBoolean` + `AtomicReference<Thread>` for prompt,
 idempotent close/cancellation signalling
 * `Flow.Publisher` adaptation (`FlowPublisherAdapter`) — `AtomicReference<Subscription>` with CAS-guarded
-  onSubscribe (§2.5 compliance), `AtomicBoolean` done-guard for single-value path, and close-aware queue signalling
+  onSubscribe (§2.5 compliance), `AtomicBoolean` done-guard for single-value path, and close-aware
+  queue signalling; all signals (`onNext`/`onError`/`onComplete`) use blocking `put()` for
+  guaranteed delivery; demand is signalled before the consumer's awaitable completes to prevent livelock
 * Defer scopes — per-task `ArrayDeque`, confined to a single thread (no sharing)
 * `DELAY_SCHEDULER` — single daemon thread for non-blocking timer operations
 
@@ -1131,7 +1154,8 @@ JavaScript, C#, Kotlin, and Swift, for developers familiar with those languages.
 | task priority / executor chosen by runtime
 
 | **Third-party support**
-| `AwaitableAdapterRegistry` SPI
+| `AwaitableAdapterRegistry` SPI +
+`Awaitable.from()` / `AsyncStream.from()`
 | _(native `thenable` protocol)_
 | Custom awaiters via `GetAwaiter()`
 | coroutine adapters / bridges
@@ -1195,6 +1219,9 @@ JavaScript, C#, Kotlin, and Swift, for developers familiar with those languages.
 | Pre-computed result
 | `Awaitable.of(value)` / `Awaitable.failed(error)`
 
+| Type conversion
+| `Awaitable.from(source)` / `AsyncStream.from(source)`
+
 | Annotation form
 | `@Async def methodName() { ... }`