GROOVY-9381: Add native async/await support

Author: daniellansun

build.gradle

@@ -122,6 +122,8 @@ dependencies {
     testImplementation "com.thoughtworks.qdox:qdox:${versions.qdox}"
     testImplementation "com.fasterxml.jackson.core:jackson-databind:${versions.jackson}"
     testImplementation "com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:${versions.jackson}"
+    testImplementation "io.reactivex.rxjava3:rxjava:${versions.rxjava3}"
+    testImplementation "io.projectreactor:reactor-core:${versions.reactor}"
 
     testFixturesImplementation projects.groovyXml
     testFixturesImplementation projects.groovyTest

gradle/verification-metadata.xml

@@ -114,6 +114,7 @@
          <ignored-key id="C71FB765CD9DE313" reason="Key couldn't be downloaded from any key server"/>
          <ignored-key id="C7CA19B7B620D787" reason="Key couldn't be downloaded from any key server"/>
          <ignored-key id="CA80D1F0EB6CA4BA" reason="Key couldn't be downloaded from any key server"/>
+         <ignored-key id="D1031D14464180E0" reason="Key couldn't be downloaded from any key server"/>
          <ignored-key id="D2151178A123C97F" reason="Key couldn't be downloaded from any key server"/>
          <ignored-key id="D364ABAA39A47320" reason="Key couldn't be downloaded from any key server"/>
          <ignored-key id="D7742D58455ECC7C" reason="Key couldn't be downloaded from any key server"/>
@@ -861,6 +862,16 @@
             <sha512 value="f220e44fe6b61f8dbb61226f832dfb16a09584384540fd48a4dff5c4de9fee060623f85cbead720dfe776aa25105949e70758a9bb1d9db43f63068d8d22164c9" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="io.projectreactor" name="reactor-core" version="3.7.3">
+         <artifact name="reactor-core-3.7.3.jar">
+            <pgp value="48B086A7D843CFA258E83286928FBF39003C0425"/>
+         </artifact>
+      </component>
+      <component group="io.reactivex.rxjava3" name="rxjava" version="3.1.10">
+         <artifact name="rxjava-3.1.10.jar">
+            <pgp value="E9CC3CD1AE59E851E4DB3FA350FFD7487D34B5B9"/>
+         </artifact>
+      </component>
       <component group="jakarta.activation" name="jakarta.activation-api" version="1.2.1">
          <artifact name="jakarta.activation-api-1.2.1.jar">
             <pgp value="6DD3B8C64EF75253BEB2C53AD908A43FB7EC07AC"/>
@@ -2216,6 +2227,11 @@
             <sha512 value="adcc480f68828ffd68d03846be852988b595c2e1bb69224d273578dd6c2ad2773edfe96625a7c00bc40ae0f2d1cac8412eaa54b88cc8e681b0b4c0ee3b082333" origin="Generated by Gradle"/>
          </artifact>
       </component>
+      <component group="org.reactivestreams" name="reactive-streams" version="1.0.4">
+         <artifact name="reactive-streams-1.0.4.jar">
+            <sha512 value="cdab6bd156f39106cd6bbfd47df1f4b0a89dc4aa28c68c31ef12a463193c688897e415f01b8d7f0d487b0e6b5bd2f19044bf8605704b024f26d6aa1f4f9a2471" origin="Generated by Gradle" reason="A key couldn't be downloaded"/>
+         </artifact>
+      </component>
       <component group="org.reflections" name="reflections" version="0.10.2">
          <artifact name="reflections-0.10.2.jar">
             <pgp value="3F2A008A91D11A7FAC4A0786F13D3E721D56BD54"/>

src/antlr/GroovyLexer.g4

@@ -392,6 +392,9 @@ DEF             : 'def';
 IN              : 'in';
 TRAIT           : 'trait';
 THREADSAFE      : 'threadsafe'; // reserved keyword
+ASYNC           : 'async';
+AWAIT           : 'await';
+DEFER           : 'defer';
 
 // §3.9 Keywords
 BuiltInPrimitiveType

src/antlr/GroovyParser.g4

@@ -134,6 +134,7 @@ modifier
           |   VOLATILE
           |   DEF
           |   VAR
+          |   ASYNC
           )
     ;
 
@@ -600,7 +601,7 @@ switchStatement
     ;
 
 loopStatement
-    :   FOR LPAREN forControl RPAREN nls statement                                                            #forStmtAlt
+    :   FOR AWAIT? LPAREN forControl RPAREN nls statement                                                     #forStmtAlt
     |   WHILE expressionInPar nls statement                                                                   #whileStmtAlt
     |   DO nls statement nls WHILE expressionInPar                                                            #doWhileStmtAlt
     ;
@@ -642,6 +643,8 @@ statement
     |   continueStatement                                                                                   #continueStmtAlt
     |   { inSwitchExpressionLevel > 0 }?
         yieldStatement                                                                                      #yieldStmtAlt
+    |   YIELD RETURN nls expression                                                                         #yieldReturnStmtAlt
+    |   DEFER nls statementExpression                                                                       #deferStmtAlt
     |   identifier COLON nls statement                                                                      #labeledStmtAlt
     |   assertStatement                                                                                     #assertStmtAlt
     |   localVariableDeclaration                                                                            #localVariableDeclarationStmtAlt
@@ -778,6 +781,10 @@ expression
     // must come before postfixExpression to resolve the ambiguities between casting and call on parentheses expression, e.g. (int)(1 / 2)
     :   castParExpression castOperandExpression                                             #castExprAlt
 
+    // async closure/lambda must come before postfixExpression to resolve the ambiguities between async and method call, e.g. async { ... }
+    |   ASYNC nls closureOrLambdaExpression                                                 #asyncClosureExprAlt
+    |   AWAIT nls (LPAREN expression RPAREN | expression)                                   #awaitExprAlt
+
     // qualified names, array expressions, method invocation, post inc/dec
     |   postfixExpression                                                                   #postfixExprAlt
 
@@ -1228,6 +1235,9 @@ identifier
     :   Identifier
     |   CapitalizedIdentifier
     |   AS
+    |   ASYNC
+    |   AWAIT
+    |   DEFER
     |   IN
     |   PERMITS
     |   RECORD
@@ -1246,12 +1256,15 @@ keywords
     :   ABSTRACT
     |   AS
     |   ASSERT
+    |   ASYNC
+    |   AWAIT
     |   BREAK
     |   CASE
     |   CATCH
     |   CLASS
     |   CONST
     |   CONTINUE
+    |   DEFER
     |   DEF
     |   DEFAULT
     |   DO

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

@@ -0,0 +1,97 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ */
+package groovy.concurrent;
+
+/**
+ * Asynchronous iteration abstraction, analogous to C#'s
+ * {@code IAsyncEnumerable<T>} or JavaScript's async iterables.
+ * <p>
+ * Used with the {@code for await} syntax:
+ * <pre>
+ * for await (item in asyncStream) {
+ *     process(item)
+ * }
+ * </pre>
+ * <p>
+ * An {@code AsyncStream} can be produced in several ways:
+ * <ul>
+ *   <li>Using {@code yield return} inside an {@code async} method or closure
+ *       to create a generator-style stream</li>
+ *   <li>Adapting JDK {@link java.util.concurrent.Flow.Publisher} instances
+ *       (supported out of the box by the built-in adapter)</li>
+ *   <li>Adapting third-party reactive types (Reactor {@code Flux}, RxJava
+ *       {@code Observable}) via {@link AwaitableAdapter}</li>
+ * </ul>
+ *
+ * @param <T> the element type
+ * @see AwaitableAdapter
+ * @see AwaitableAdapterRegistry
+ * @since 6.0.0
+ */
+public interface AsyncStream<T> extends AutoCloseable {
+
+    /**
+     * Asynchronously advances to the next element. Returns an {@link Awaitable}
+     * that completes with {@code true} if an element is available, or
+     * {@code false} if the stream is exhausted.
+     */
+    Awaitable<Boolean> moveNext();
+
+    /**
+     * Returns the current element. Must only be called after {@link #moveNext()}
+     * has completed with {@code true}.
+     */
+    T getCurrent();
+
+    /**
+     * Closes the stream and releases any associated resources.
+     * <p>
+     * The default implementation is a no-op.  Implementations that bridge to
+     * generators, publishers, or other resource-owning sources may override
+     * this to propagate cancellation upstream.  Compiler-generated
+     * {@code for await} loops invoke {@code close()} automatically from a
+     * {@code finally} block, including on early {@code break}, {@code return},
+     * and exceptional exit.
+     *
+     * @since 6.0.0
+     */
+    @Override
+    default void close() {
+    }
+
+    /**
+     * Returns an empty {@code AsyncStream} that completes immediately.
+     */
+    @SuppressWarnings("unchecked")
+    static <T> AsyncStream<T> empty() {
+        return (AsyncStream<T>) EMPTY;
+    }
+
+    /**
+     * Singleton empty stream instance.
+     * <p>
+     * This is an implementation detail backing {@link #empty()}.
+     * User code should call {@code AsyncStream.empty()} rather than
+     * referencing this field directly.
+     */
+    AsyncStream<Object> EMPTY = new AsyncStream<>() {
+        @Override public Awaitable<Boolean> moveNext() { return Awaitable.of(false); }
+        @Override public Object getCurrent() { return null; }
+    };
+}

src/main/java/groovy/concurrent/AwaitResult.java

@@ -0,0 +1,122 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ */
+package groovy.concurrent;
+
+import java.util.Objects;
+import java.util.function.Function;
+
+/**
+ * Represents the outcome of an asynchronous computation that may have
+ * succeeded or failed. This is used by {@code awaitAllSettled()} —
+ * the Groovy equivalent of JavaScript's {@code Promise.allSettled()}.
+ * <p>
+ * An {@code AwaitResult} is either a {@linkplain #isSuccess() success}
+ * carrying a value, or a {@linkplain #isFailure() failure} carrying a
+ * {@link Throwable}.
+ *
+ * @param <T> the value type
+ * @since 6.0.0
+ */
+public final class AwaitResult<T> {
+
+    private final T value;
+    private final Throwable error;
+    private final boolean success;
+
+    private AwaitResult(T value, Throwable error, boolean success) {
+        this.value = value;
+        this.error = error;
+        this.success = success;
+    }
+
+    /**
+     * Creates a successful result with the given value.
+     *
+     * @param value the computation result (may be {@code null})
+     * @param <T>   the value type
+     * @return a success result wrapping the value
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> AwaitResult<T> success(Object value) {
+        return new AwaitResult<>((T) value, null, true);
+    }
+
+    /**
+     * Creates a failure result with the given exception.
+     *
+     * @param error the exception that caused the failure; must not be {@code null}
+     * @param <T>   the value type (never actually used, since the result is a failure)
+     * @return a failure result wrapping the exception
+     * @throws NullPointerException if {@code error} is {@code null}
+     */
+    public static <T> AwaitResult<T> failure(Throwable error) {
+        return new AwaitResult<>(null, Objects.requireNonNull(error), false);
+    }
+
+    /** Returns {@code true} if this result represents a successful completion. */
+    public boolean isSuccess() {
+        return success;
+    }
+
+    /** Returns {@code true} if this result represents a failed completion. */
+    public boolean isFailure() {
+        return !success;
+    }
+
+    /**
+     * Returns the value if successful.
+     *
+     * @return the computation result
+     * @throws IllegalStateException if this result represents a failure
+     */
+    public T getValue() {
+        if (!success) throw new IllegalStateException("Cannot get value from a failed result");
+        return value;
+    }
+
+    /**
+     * Returns the exception if failed.
+     *
+     * @return the exception that caused the failure
+     * @throws IllegalStateException if this result represents a success
+     */
+    public Throwable getError() {
+        if (success) throw new IllegalStateException("Cannot get error from a successful result");
+        return error;
+    }
+
+    /**
+     * Returns the value if successful, or applies the given function to
+     * the error to produce a fallback value.
+     */
+    public T getOrElse(Function<Throwable, ? extends T> fallback) {
+        return success ? value : fallback.apply(error);
+    }
+
+    /**
+     * Returns a human-readable representation of this result:
+     * {@code AwaitResult.Success[value]} or {@code AwaitResult.Failure[error]}.
+     */
+    @Override
+    public String toString() {
+        return success
+            ? "AwaitResult.Success[" + value + "]"
+            : "AwaitResult.Failure[" + error + "]";
+    }
+}