From 9d50409a09cecb77f4d47589f50d2965073da4f9 Mon Sep 17 00:00:00 2001 From: Daniel Sun Date: Tue, 21 Jul 2026 00:58:40 +0900 Subject: [PATCH] GROOVY-12181: Refactor async runtime and AST helpers introduced by GROOVY-9381 --- .../java/groovy/concurrent/AwaitResult.java | 5 +- .../java/groovy/concurrent/Awaitable.java | 7 +- .../java/groovy/concurrent/package-info.java | 17 +- .../groovy/parser/antlr4/AstBuilder.java | 49 +-- .../groovy/runtime/async/AsyncExecutors.java | 128 ++++++++ .../groovy/runtime/async/AsyncSupport.java | 282 ++++-------------- .../runtime/async/AwaitCombinators.java | 201 +++++++++++++ .../groovy/runtime/async/GroovyPromise.java | 9 +- .../groovy/runtime/async/package-info.java | 20 +- .../transform/AsyncTransformHelper.java | 127 +++++--- src/spec/doc/core-async-await.adoc | 4 +- src/test/groovy/groovy/AsyncAwaitTest.groovy | 40 +++ .../groovy/concurrent/AwaitResultTest.groovy | 118 ++++++++ .../async/AsyncSupportInternalsTest.groovy | 259 ++++++++++++++++ .../transform/AsyncTransformHelperTest.groovy | 184 ++++++++++++ 15 files changed, 1131 insertions(+), 319 deletions(-) create mode 100644 src/main/java/org/apache/groovy/runtime/async/AsyncExecutors.java create mode 100644 src/main/java/org/apache/groovy/runtime/async/AwaitCombinators.java create mode 100644 src/test/groovy/groovy/concurrent/AwaitResultTest.groovy create mode 100644 src/test/groovy/org/apache/groovy/runtime/async/AsyncSupportInternalsTest.groovy create mode 100644 src/test/groovy/org/codehaus/groovy/transform/AsyncTransformHelperTest.groovy diff --git a/src/main/java/groovy/concurrent/AwaitResult.java b/src/main/java/groovy/concurrent/AwaitResult.java index f7f77879b90..eed54a524cf 100644 --- a/src/main/java/groovy/concurrent/AwaitResult.java +++ b/src/main/java/groovy/concurrent/AwaitResult.java @@ -69,9 +69,8 @@ private AwaitResult(T value, Throwable error, boolean success) { * @param the value type * @return a success result wrapping the value */ - @SuppressWarnings("unchecked") - public static AwaitResult success(Object value) { - return new AwaitResult<>((T) value, null, true); + public static AwaitResult success(T value) { + return new AwaitResult<>(value, null, true); } /** diff --git a/src/main/java/groovy/concurrent/Awaitable.java b/src/main/java/groovy/concurrent/Awaitable.java index 59f06f08764..bb941cd1e7f 100644 --- a/src/main/java/groovy/concurrent/Awaitable.java +++ b/src/main/java/groovy/concurrent/Awaitable.java @@ -439,9 +439,10 @@ static Awaitable any(Object... sources) { * Returns an {@code Awaitable} that completes with the result of the first * source that succeeds. Individual failures are silently absorbed; only * when all sources have failed does the returned awaitable reject - * with an {@link IllegalStateException} whose - * {@linkplain Throwable#getSuppressed() suppressed} array contains every - * individual error. + * with an aggregate {@link java.util.concurrent.CompletionException} + * (message {@code "All N tasks failed"}, cause = first failure, remaining + * failures as suppressed). Under {@code await} exception transparency the + * cause is rethrown. *

* This is the Groovy equivalent of JavaScript's {@code Promise.any()}. * Contrast with {@link #any(Object...)} which returns the first result to diff --git a/src/main/java/groovy/concurrent/package-info.java b/src/main/java/groovy/concurrent/package-info.java index 1ffb533d6d8..950b8906192 100644 --- a/src/main/java/groovy/concurrent/package-info.java +++ b/src/main/java/groovy/concurrent/package-info.java @@ -21,10 +21,23 @@ * Structured concurrency abstractions for asynchronous and parallel programming. * *

- * Provides high-level concurrency primitives: {@link groovy.concurrent.Actor} (message-passing), - * {@link groovy.concurrent.AsyncScope} (structured concurrency), {@link groovy.concurrent.Pool} (thread management), + * Language-facing async types: + *

    + *
  • {@link groovy.concurrent.Awaitable} — async computation handle and + * static combinator surface used with the {@code await} keyword
    • + *
  • {@link groovy.concurrent.AwaitResult} — success/failure value from + * {@link groovy.concurrent.Awaitable#allSettled(Object...)}
    • + *
  • {@link groovy.concurrent.AsyncScope} — structured concurrency
    • + *
  • {@link groovy.concurrent.AsyncChannel} — CSP-style async channel
    • + *
  • {@link groovy.concurrent.AwaitableAdapter} / + * {@link groovy.concurrent.AwaitableAdapterRegistry} — SPI for + * third-party async types (RxJava, Reactor, …)
    • + *
+ * Additional high-level primitives: {@link groovy.concurrent.Actor} + * (message-passing), {@link groovy.concurrent.Pool} (thread management), * and {@link groovy.concurrent.ParallelScope} (combined async+parallel). * Supports virtual threads (JDK 21+) and fail-fast exception handling. + * Runtime implementation details live in {@code org.apache.groovy.runtime.async}. *

*/ package groovy.concurrent; 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 deaaf9988d5..a35a61c7916 100644 --- a/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java +++ b/src/main/java/org/apache/groovy/parser/antlr4/AstBuilder.java @@ -539,20 +539,7 @@ public Statement visitForStmtAlt(final ForStmtAltContext ctx) { visitAnnotationsOpt(ctx.annotationsOpt()).forEach(forStatement::addStatementAnnotation); if (isForAwait) { - // Transform collection expression: wrap in AsyncSupport.toIterable() - // and wrap the loop in try/finally to ensure cleanup on break/exception - Expression original = forStatement.getCollectionExpression(); - String tempVar = "__forAwaitSource" + ctx.hashCode(); - Expression toIterableCall = AsyncTransformHelper.buildToIterableCall(original); - - // var $temp = AsyncSupport.toIterable(original) - Statement declStmt = stmt(declX(varX(tempVar), toIterableCall)); - forStatement.setCollectionExpression(varX(tempVar)); - - // try { for (...) { body } } finally { AsyncSupport.closeIterable($temp) } - Statement finallyStmt = stmt(AsyncTransformHelper.buildCloseIterableCall(varX(tempVar))); - TryCatchStatement tryCatch = new TryCatchStatement(forStatement, finallyStmt); - return configureAST(block(declStmt, tryCatch), ctx); + return configureAST(AsyncTransformHelper.wrapForAwaitLoop(forStatement), ctx); } return forStatement; @@ -3090,39 +3077,7 @@ public Expression visitAwaitExprAlt(final AwaitExprAltContext ctx) { @Override public Expression visitAsyncClosureExprAlt(final AsyncClosureExprAltContext ctx) { ClosureExpression closure = this.visitClosureOrLambdaExpression(ctx.closureOrLambdaExpression()); - boolean hasYieldReturn = AsyncTransformHelper.containsYieldReturn(closure.getCode()); - boolean hasDefer = AsyncTransformHelper.containsDefer(closure.getCode()); - - if (hasDefer) { - Statement wrappedBody = AsyncTransformHelper.wrapWithDeferScope(closure.getCode()); - ClosureExpression newClosure = new ClosureExpression(closure.getParameters(), wrappedBody); - newClosure.setVariableScope(closure.getVariableScope()); - newClosure.setSourcePosition(closure); - closure = newClosure; - } - - if (hasYieldReturn) { - // Inject synthetic $__asyncGen__ as first parameter - Parameter genParam = AsyncTransformHelper.createGenParam(); - Parameter[] existingParams = closure.getParameters(); - boolean hasUserParams = existingParams != null && existingParams.length > 0; - Parameter[] newParams; - if (hasUserParams) { - newParams = new Parameter[existingParams.length + 1]; - newParams[0] = genParam; - System.arraycopy(existingParams, 0, newParams, 1, existingParams.length); - } else { - newParams = new Parameter[]{genParam}; - } - ClosureExpression genClosure = new ClosureExpression(newParams, closure.getCode()); - genClosure.setVariableScope(closure.getVariableScope()); - genClosure.setSourcePosition(closure); - return configureAST(AsyncTransformHelper.buildAsyncGeneratorCall( - new ArgumentListExpression(genClosure)), ctx); - } else { - return configureAST(AsyncTransformHelper.buildAsyncCall( - new ArgumentListExpression(closure)), ctx); - } + return configureAST(AsyncTransformHelper.transformAsyncClosure(closure), ctx); } @Override diff --git a/src/main/java/org/apache/groovy/runtime/async/AsyncExecutors.java b/src/main/java/org/apache/groovy/runtime/async/AsyncExecutors.java new file mode 100644 index 00000000000..8114747b176 --- /dev/null +++ b/src/main/java/org/apache/groovy/runtime/async/AsyncExecutors.java @@ -0,0 +1,128 @@ +/* + * 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 org.apache.groovy.runtime.async; + +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Executor and scheduler configuration for the async runtime. + *

+ * Package-private implementation detail of {@link AsyncSupport}. On JDK 21+ + * the default executor is a virtual-thread-per-task executor; earlier JDKs use a + * bounded cached daemon pool sized by {@code groovy.async.parallelism}. + * + * @since 6.0.0 + */ +final class AsyncExecutors { + + private static final boolean VIRTUAL_THREADS_AVAILABLE; + private static final Executor VIRTUAL_THREAD_EXECUTOR; + private static final Executor FALLBACK_EXECUTOR; + + static { + Executor vtExecutor = null; + boolean vtAvailable = false; + try { + MethodHandle mh = MethodHandles.lookup().findStatic( + Executors.class, "newVirtualThreadPerTaskExecutor", + MethodType.methodType(ExecutorService.class)); + vtExecutor = (Executor) mh.invoke(); + vtAvailable = true; + } catch (Throwable ignored) { + // JDK < 21 — virtual threads not available + } + VIRTUAL_THREAD_EXECUTOR = vtExecutor; + VIRTUAL_THREADS_AVAILABLE = vtAvailable; + + int maxThreads = getIntegerSafe("groovy.async.parallelism", 256); + if (!VIRTUAL_THREADS_AVAILABLE) { + FALLBACK_EXECUTOR = new ThreadPoolExecutor( + 0, maxThreads, + 60L, TimeUnit.SECONDS, + new SynchronousQueue<>(), + r -> { + Thread t = new Thread(r); + t.setDaemon(true); + @SuppressWarnings("deprecation") + long id = t.getId(); + t.setName("groovy-async-" + id); + return t; + }, + new ThreadPoolExecutor.CallerRunsPolicy()); + } else { + FALLBACK_EXECUTOR = null; + } + } + + private static volatile Executor defaultExecutor = createDefaultExecutor(); + + private static final ScheduledExecutorService SCHEDULER = + Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "groovy-async-scheduler"); + t.setDaemon(true); + return t; + }); + + private AsyncExecutors() { } + + static boolean isVirtualThreadsAvailable() { + return VIRTUAL_THREADS_AVAILABLE; + } + + static Executor getExecutor() { + return defaultExecutor; + } + + /** + * Sets the executor used for async tasks. Passing {@code null} restores the + * platform default (virtual threads on JDK 21+, cached pool otherwise). + */ + static void setExecutor(Executor executor) { + defaultExecutor = executor != null ? executor : createDefaultExecutor(); + } + + static void resetExecutor() { + defaultExecutor = createDefaultExecutor(); + } + + static ScheduledExecutorService getScheduler() { + return SCHEDULER; + } + + private static Executor createDefaultExecutor() { + return VIRTUAL_THREADS_AVAILABLE ? VIRTUAL_THREAD_EXECUTOR : FALLBACK_EXECUTOR; + } + + private static int getIntegerSafe(String name, int defaultValue) { + try { + return Integer.getInteger(name, defaultValue); + } catch (SecurityException ignore) { + return defaultValue; + } + } +} 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 588f5a30845..6c78b0990bc 100644 --- a/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java +++ b/src/main/java/org/apache/groovy/runtime/async/AsyncSupport.java @@ -23,13 +23,9 @@ import groovy.concurrent.AwaitableAdapterRegistry; import java.io.Closeable; -import java.lang.invoke.MethodHandle; -import java.lang.invoke.MethodHandles; -import java.lang.invoke.MethodType; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.UndeclaredThrowableException; import java.util.ArrayDeque; -import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Deque; @@ -43,26 +39,22 @@ import java.util.concurrent.CompletionStage; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.SynchronousQueue; -import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.function.Supplier; /** * Internal runtime support for the {@code async}/{@code await}/{@code defer} language features. *

- * This class contains the actual implementation invoked by compiler-generated code. - * User code should prefer the static methods on {@link groovy.concurrent.Awaitable} - * for combinators and configuration. + * This class is the entry point invoked by compiler-generated code and also + * exposes the combinator and configuration surface used by + * {@link groovy.concurrent.Awaitable}. Combinator algorithms live in + * {@link AwaitCombinators}; executor/scheduler configuration lives in + * {@link AsyncExecutors}. *

* Thread pool configuration: *

    @@ -71,7 +63,8 @@ *
  • On JDK 17-20 the fallback is a cached daemon thread pool * whose maximum size is controlled by the system property * {@code groovy.async.parallelism} (default: {@code 256}).
    • - *
  • The executor can be overridden at any time via {@link #setExecutor}.
    • + *
  • The executor can be overridden via {@link #setExecutor}; pass + * {@code null} to restore the platform default.
    • *
*

* Exception handling follows a transparency principle: the @@ -82,82 +75,42 @@ */ public class AsyncSupport { - private static final boolean VIRTUAL_THREADS_AVAILABLE; - private static final Executor VIRTUAL_THREAD_EXECUTOR; - private static final int FALLBACK_MAX_THREADS; - private static final Executor FALLBACK_EXECUTOR; - - static { - Executor vtExecutor = null; - boolean vtAvailable = false; - try { - MethodHandle mh = MethodHandles.lookup().findStatic( - Executors.class, "newVirtualThreadPerTaskExecutor", - MethodType.methodType(ExecutorService.class)); - vtExecutor = (Executor) mh.invoke(); - vtAvailable = true; - } catch (Throwable ignored) { - // JDK < 21 — virtual threads not available - } - VIRTUAL_THREAD_EXECUTOR = vtExecutor; - VIRTUAL_THREADS_AVAILABLE = vtAvailable; - - FALLBACK_MAX_THREADS = getIntegerSafe("groovy.async.parallelism", 256); - if (!VIRTUAL_THREADS_AVAILABLE) { - FALLBACK_EXECUTOR = new ThreadPoolExecutor( - 0, FALLBACK_MAX_THREADS, - 60L, TimeUnit.SECONDS, - new SynchronousQueue<>(), - r -> { - Thread t = new Thread(r); - t.setDaemon(true); - @SuppressWarnings("deprecation") - long id = t.getId(); - t.setName("groovy-async-" + id); - return t; - }, - new ThreadPoolExecutor.CallerRunsPolicy()); - } else { - FALLBACK_EXECUTOR = null; - } - } - - private static volatile Executor defaultExecutor = createDefaultExecutor(); - - private static final ScheduledExecutorService SCHEDULER = - Executors.newSingleThreadScheduledExecutor(r -> { - Thread t = new Thread(r, "groovy-async-scheduler"); - t.setDaemon(true); - return t; - }); - private AsyncSupport() { } // ---- executor configuration ----------------------------------------- /** Returns the shared scheduler for delays, timeouts, and scope deadlines. */ public static ScheduledExecutorService getScheduler() { - return SCHEDULER; + return AsyncExecutors.getScheduler(); } /** Returns {@code true} if running on JDK 21+ with virtual thread support. */ public static boolean isVirtualThreadsAvailable() { - return VIRTUAL_THREADS_AVAILABLE; + return AsyncExecutors.isVirtualThreadsAvailable(); } /** Returns the current executor used for async tasks. */ public static Executor getExecutor() { - return defaultExecutor; + return AsyncExecutors.getExecutor(); } - /** Sets the executor used for async tasks. */ + /** + * Sets the executor used for async tasks. + *

+ * Pass {@code null} to restore the platform default (virtual threads on + * JDK 21+, cached daemon pool otherwise). The change takes effect + * for subsequent {@code async} launches; in-flight tasks keep the + * executor that started them. + * + * @param executor the executor to use, or {@code null} to reset + */ public static void setExecutor(Executor executor) { - defaultExecutor = Objects.requireNonNull(executor, "executor must not be null"); + AsyncExecutors.setExecutor(executor); } - /** Resets the executor to the default (virtual threads on JDK 21+, cached pool otherwise). */ + /** Resets the executor to the platform default. Equivalent to {@code setExecutor(null)}. */ public static void resetExecutor() { - defaultExecutor = createDefaultExecutor(); + AsyncExecutors.resetExecutor(); } // ---- await overloads ------------------------------------------------ @@ -211,7 +164,12 @@ public static T await(Future future) { /** * Awaits an arbitrary object by adapting it via {@link Awaitable#from(Object)}. - * This is the fallback overload called by compiler-generated await expressions. + *

+ * This is the fallback overload called by compiler-generated {@code await} + * expressions. The compiler inserts a cast to {@link Object} so that + * overload resolution does not have to choose among {@code Awaitable}, + * {@code CompletionStage}, and {@code Future} when a value implements more + * than one of them (e.g. {@link CompletableFuture}). */ @SuppressWarnings("unchecked") public static T await(Object source) { @@ -231,7 +189,7 @@ public static T await(Object source) { */ public static Awaitable executeAsync(Supplier supplier, Executor executor) { Objects.requireNonNull(supplier, "supplier must not be null"); - Executor targetExecutor = executor != null ? executor : defaultExecutor; + Executor targetExecutor = executor != null ? executor : AsyncExecutors.getExecutor(); return GroovyPromise.of(CompletableFuture.supplyAsync(() -> { try { return supplier.get(); @@ -245,11 +203,12 @@ public static Awaitable executeAsync(Supplier supplier, Executor execu * Executes the given supplier asynchronously using the default executor. */ public static Awaitable async(Supplier supplier) { - return executeAsync(supplier, defaultExecutor); + return executeAsync(supplier, AsyncExecutors.getExecutor()); } /** - * Lightweight task spawn. Executes the supplier asynchronously using the default executor. + * Lightweight task spawn. Alias of {@link #async(Supplier)} for Go-style + * ergonomics; semantics are identical. */ public static Awaitable go(Supplier supplier) { return async(supplier); @@ -286,7 +245,6 @@ public static void defer(Deque> scope, * subsequent exceptions are added as suppressed. If a deferred action returns * a Future/Awaitable, the result is awaited before continuing. */ - @SuppressWarnings("unchecked") public static void executeDeferScope(Deque> scope) { if (scope == null || scope.isEmpty()) return; Throwable firstError = null; @@ -349,11 +307,10 @@ public static void yieldReturn(Object bridge, Object value) { * @param the element type * @return an Iterable that yields values from the generator */ - @SuppressWarnings("unchecked") public static Iterable asyncGenerator(Consumer body) { Objects.requireNonNull(body, "body must not be null"); GeneratorBridge bridge = new GeneratorBridge<>(); - defaultExecutor.execute(() -> { + AsyncExecutors.getExecutor().execute(() -> { try { body.accept(bridge); bridge.complete(); @@ -387,14 +344,15 @@ public static Iterable toIterable(Object source) { return () -> iter; } if (source instanceof Object[]) return (Iterable) Arrays.asList((Object[]) source); - // Try adapter registry return AwaitableAdapterRegistry.toIterable(source); } /** * Closes a source if it implements {@link Closeable} or * {@link AutoCloseable}. Called by compiler-generated finally block - * in {@code for await} loops. + * in {@code for await} loops. Cleanup exceptions are swallowed so they + * cannot mask the original loop error; prefer robust {@code close()} + * implementations. */ public static void closeIterable(Object source) { if (source instanceof Closeable c) { @@ -410,31 +368,31 @@ public static void closeIterable(Object source) { } } - // ---- combinators ---------------------------------------------------- + // ---- combinators (delegate to AwaitCombinators) --------------------- /** * Waits for all given sources to complete, returning their results in order. - * Multi-arg {@code await(a, b, c)} desugars to this. + * Multi-arg {@code await(a, b, c)} desugars to the non-blocking + * {@link #allAsync(Object...)} form, then awaits it. */ public static List all(Object... sources) { - CompletableFuture[] futures = toCombinatorFutures(sources); - CompletableFuture.allOf(futures).join(); - return getJoinedResults(futures); + return AwaitCombinators.all(sources); } /** * Returns the result of the first source to complete (success or failure). */ public static T any(Object... sources) { - return AsyncSupport.await(AsyncSupport.anyAsync(sources).toCompletableFuture()); + return AwaitCombinators.any(sources); } /** * Returns the result of the first source to complete successfully. - * Only fails when all sources fail. + * Only fails when all sources fail (aggregate {@link CompletionException}; + * {@code await} transparency rethrows the cause). */ public static T first(Object... sources) { - return AsyncSupport.await(AsyncSupport.firstAsync(sources).toCompletableFuture()); + return AwaitCombinators.first(sources); } /** @@ -442,136 +400,27 @@ public static T first(Object... sources) { * {@link AwaitResult} without throwing. */ public static List> allSettled(Object... sources) { - return AsyncSupport.await(allSettledAsync(sources).toCompletableFuture()); + return AwaitCombinators.allSettled(sources); } - // ---- async combinator variants (return Awaitable, non-blocking) ------ - /** Non-blocking variant of {@link #all} — returns an Awaitable. */ public static Awaitable> allAsync(Object... sources) { - CompletableFuture[] futures = toCombinatorFutures(sources); - - // allOf is naturally fail-fast: it completes as soon as any - // source fails OR all sources succeed. We track the first - // failure explicitly because allOf doesn't guarantee which - // exception propagates when multiple futures fail. - var firstError = new AtomicReference(); - for (CompletableFuture f : futures) { - f.whenComplete((v, e) -> { - if (e != null) firstError.compareAndSet(null, e); - }); - } - - CompletableFuture> combined = CompletableFuture.allOf(futures) - .thenApply(v -> getJoinedResults(futures)); - - // Replace allOf's arbitrary exception with the temporally-first one - CompletableFuture> withFirstError = combined.exceptionally(e -> { - Throwable first = firstError.get(); - if (first != null && first != e && first != e.getCause()) { - throw first instanceof CompletionException ce ? ce : new CompletionException(first); - } - throw e instanceof CompletionException ce ? ce : new CompletionException(e); - }); - return GroovyPromise.of(withFirstError); + return AwaitCombinators.allAsync(sources); } /** Non-blocking variant of {@link #any} — returns an Awaitable. */ - @SuppressWarnings("unchecked") public static Awaitable anyAsync(Object... sources) { - return (Awaitable) GroovyPromise.of(CompletableFuture.anyOf(toCombinatorFutures(sources))); + return AwaitCombinators.anyAsync(sources); } /** Non-blocking variant of {@link #first} — returns an Awaitable. */ - @SuppressWarnings("unchecked") public static Awaitable firstAsync(Object... sources) { - CompletableFuture[] futures = toFirstCombinatorFutures(sources); - CompletableFuture result = new CompletableFuture<>(); - var remainingFailures = new AtomicInteger(futures.length); - List errors = Collections.synchronizedList(new ArrayList<>(futures.length)); - for (CompletableFuture future : futures) { - future.whenComplete((value, error) -> { - if (error == null) { - result.complete(value); - return; - } - errors.add(error); - if (remainingFailures.decrementAndGet() == 0) { - result.completeExceptionally(aggregateFirstFailures(futures.length, errors)); - } - }); - } - return GroovyPromise.of(result); + return AwaitCombinators.firstAsync(sources); } /** Non-blocking variant of {@link #allSettled} — returns an Awaitable. */ public static Awaitable>> allSettledAsync(Object... sources) { - CompletableFuture[] futures = toCombinatorFutures(sources); - CompletableFuture>> combined = CompletableFuture.allOf( - Arrays.stream(futures) - .map(future -> future.handle((value, error) -> null)) - .toArray(CompletableFuture[]::new) - ) - .thenApply(v -> getAwaitResults(futures)); - return GroovyPromise.of(combined); - } - - @SuppressWarnings("unchecked") - private static List getJoinedResults(CompletableFuture[] futures) { - List results = new ArrayList<>(futures.length); - for (CompletableFuture future : futures) { - results.add((T) future.join()); - } - return results; - } - - private static List> getAwaitResults(CompletableFuture[] futures) { - List> results = new ArrayList<>(futures.length); - for (CompletableFuture f : futures) { - try { - results.add(AwaitResult.success(f.join())); - } catch (CompletionException e) { - results.add(AwaitResult.failure(unwrap(e))); - } catch (CancellationException e) { - results.add(AwaitResult.failure(e)); - } - } - return results; - } - - private static CompletableFuture[] toCombinatorFutures(Object... sources) { - return Arrays.stream(sources) - .map(source -> Awaitable.from(source).toCompletableFuture()) - .toArray(CompletableFuture[]::new); - } - - @SuppressWarnings("unchecked") - private static CompletableFuture[] toFirstCombinatorFutures(Object... sources) { - validateFirstSources(sources); - return (CompletableFuture[]) toCombinatorFutures(sources); - } - - private static void validateFirstSources(Object[] sources) { - if (sources == null) { - throw new IllegalArgumentException("sources must not be null"); - } - if (sources.length == 0) { - throw new IllegalArgumentException("sources must not be empty"); - } - for (Object source : sources) { - if (source == null) { - throw new IllegalArgumentException("sources must not contain null elements"); - } - } - } - - private static CompletionException aggregateFirstFailures(int sourceCount, List errors) { - CompletionException aggregate = new CompletionException( - "All " + sourceCount + " tasks failed", errors.get(0)); - for (int i = 1; i < errors.size(); i++) { - aggregate.addSuppressed(errors.get(i)); - } - return aggregate; + return AwaitCombinators.allSettledAsync(sources); } // ---- delay and timeout ---------------------------------------------- @@ -580,27 +429,26 @@ private static CompletionException aggregateFirstFailures(int sourceCount, List< * Returns an Awaitable that completes after the specified delay. */ public static Awaitable delay(long millis) { - CompletableFuture future = new CompletableFuture<>(); - SCHEDULER.schedule(() -> future.complete(null), millis, TimeUnit.MILLISECONDS); - return GroovyPromise.of(future); + return delay(millis, TimeUnit.MILLISECONDS); } /** Delay with explicit time unit. */ public static Awaitable delay(long duration, TimeUnit unit) { CompletableFuture future = new CompletableFuture<>(); - SCHEDULER.schedule(() -> future.complete(null), duration, unit); + AsyncExecutors.getScheduler().schedule(() -> future.complete(null), duration, unit); return GroovyPromise.of(future); } /** * Wraps a source with a timeout. If the source does not complete within - * the specified time, the returned Awaitable fails with {@link TimeoutException}. + * the specified time, the returned Awaitable fails with {@link TimeoutException} + * and the underlying computation is cancelled. */ @SuppressWarnings("unchecked") public static Awaitable orTimeout(Object source, long timeout, TimeUnit unit) { CompletableFuture future = (CompletableFuture) Awaitable.from(source).toCompletableFuture(); CompletableFuture result = new CompletableFuture<>(); - ScheduledFuture timer = SCHEDULER.schedule(() -> { + ScheduledFuture timer = AsyncExecutors.getScheduler().schedule(() -> { if (!result.isDone()) { result.completeExceptionally(new TimeoutException("Timed out after " + timeout + " " + unit)); future.cancel(true); @@ -621,12 +469,13 @@ public static Awaitable orTimeoutMillis(Object source, long millis) { /** * Wraps a source with a timeout that uses a fallback value instead of throwing. + * On timeout the underlying computation is cancelled. */ @SuppressWarnings("unchecked") public static Awaitable completeOnTimeout(Object source, T fallback, long timeout, TimeUnit unit) { CompletableFuture future = (CompletableFuture) Awaitable.from(source).toCompletableFuture(); CompletableFuture result = new CompletableFuture<>(); - ScheduledFuture timer = SCHEDULER.schedule(() -> { + ScheduledFuture timer = AsyncExecutors.getScheduler().schedule(() -> { if (!result.isDone()) { result.complete(fallback); future.cancel(true); @@ -658,6 +507,11 @@ private static RuntimeException rethrowUnwrapped(Throwable wrapper) { return null; // unreachable } + /** + * Strips JDK wrapper layers ({@link CompletionException}, + * {@link ExecutionException}, {@link InvocationTargetException}, + * {@link UndeclaredThrowableException}) to expose the original cause. + */ public static Throwable unwrap(Throwable t) { while ((t instanceof CompletionException || t instanceof ExecutionException || t instanceof InvocationTargetException @@ -675,22 +529,10 @@ private static void sneakyThrow(Throwable t) throws T { // ---- internal utilities --------------------------------------------- - private static Executor createDefaultExecutor() { - return VIRTUAL_THREADS_AVAILABLE ? VIRTUAL_THREAD_EXECUTOR : FALLBACK_EXECUTOR; - } - private static CancellationException interruptedAwait(String message, InterruptedException cause) { Thread.currentThread().interrupt(); CancellationException cancellation = new CancellationException(message); cancellation.initCause(cause); return cancellation; } - - private static int getIntegerSafe(String name, int defaultValue) { - try { - return Integer.getInteger(name, defaultValue); - } catch (SecurityException ignore) { - return defaultValue; - } - } } diff --git a/src/main/java/org/apache/groovy/runtime/async/AwaitCombinators.java b/src/main/java/org/apache/groovy/runtime/async/AwaitCombinators.java new file mode 100644 index 00000000000..9e103818a88 --- /dev/null +++ b/src/main/java/org/apache/groovy/runtime/async/AwaitCombinators.java @@ -0,0 +1,201 @@ +/* + * 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 org.apache.groovy.runtime.async; + +import groovy.concurrent.AwaitResult; +import groovy.concurrent.Awaitable; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Combinators for joining multiple async sources ({@code all}, {@code any}, + * {@code first}, {@code allSettled}). + *

+ * Package-private implementation detail of {@link AsyncSupport}. Blocking + * variants are thin wrappers over the non-blocking ones via + * {@link AsyncSupport#await(Object)}, so each combinator has a single + * implementation of its joining algorithm. + * + * @since 6.0.0 + */ +final class AwaitCombinators { + + private AwaitCombinators() { } + + // ---- blocking ------------------------------------------------------- + + @SuppressWarnings("unchecked") + static List all(Object... sources) { + return (List) AsyncSupport.await(allAsync(sources)); + } + + static T any(Object... sources) { + return AsyncSupport.await(AwaitCombinators.anyAsync(sources)); + } + + static T first(Object... sources) { + return AsyncSupport.await(AwaitCombinators.firstAsync(sources)); + } + + static List> allSettled(Object... sources) { + return AsyncSupport.await(allSettledAsync(sources)); + } + + // ---- non-blocking --------------------------------------------------- + + static Awaitable> allAsync(Object... sources) { + CompletableFuture[] futures = toFutures(sources); + + // allOf is fail-fast but does not guarantee which exception propagates + // when several futures fail; track the temporally-first error ourselves. + var firstError = new AtomicReference(); + for (CompletableFuture future : futures) { + future.whenComplete((value, error) -> { + if (error != null) { + firstError.compareAndSet(null, error); + } + }); + } + + CompletableFuture> combined = CompletableFuture.allOf(futures) + .thenApply(ignored -> getJoinedResults(futures)); + + CompletableFuture> withFirstError = combined.exceptionally(error -> { + Throwable first = firstError.get(); + if (first != null && first != error && first != error.getCause()) { + throw asCompletionException(first); + } + throw asCompletionException(error); + }); + return GroovyPromise.of(withFirstError); + } + + @SuppressWarnings("unchecked") + static Awaitable anyAsync(Object... sources) { + return (Awaitable) GroovyPromise.of(CompletableFuture.anyOf(toFutures(sources))); + } + + @SuppressWarnings("unchecked") + static Awaitable firstAsync(Object... sources) { + validateNonEmptySources(sources); + CompletableFuture[] futures = (CompletableFuture[]) toFutures(sources); + CompletableFuture result = new CompletableFuture<>(); + var remainingFailures = new AtomicInteger(futures.length); + List errors = Collections.synchronizedList(new ArrayList<>(futures.length)); + for (CompletableFuture future : futures) { + future.whenComplete((value, error) -> { + if (error == null) { + result.complete(value); + return; + } + errors.add(error); + if (remainingFailures.decrementAndGet() == 0) { + result.completeExceptionally(aggregateFirstFailures(futures.length, errors)); + } + }); + } + return GroovyPromise.of(result); + } + + static Awaitable>> allSettledAsync(Object... sources) { + CompletableFuture[] futures = toFutures(sources); + CompletableFuture>> combined = CompletableFuture.allOf( + Arrays.stream(futures) + .map(future -> future.handle((value, error) -> null)) + .toArray(CompletableFuture[]::new)) + .thenApply(ignored -> getAwaitResults(futures)); + return GroovyPromise.of(combined); + } + + // ---- helpers -------------------------------------------------------- + + private static CompletableFuture[] toFutures(Object... sources) { + if (sources == null) { + return new CompletableFuture[0]; + } + CompletableFuture[] futures = new CompletableFuture[sources.length]; + for (int i = 0; i < sources.length; i++) { + futures[i] = Awaitable.from(sources[i]).toCompletableFuture(); + } + return futures; + } + + private static void validateNonEmptySources(Object[] sources) { + if (sources == null) { + throw new IllegalArgumentException("sources must not be null"); + } + if (sources.length == 0) { + throw new IllegalArgumentException("sources must not be empty"); + } + for (Object source : sources) { + if (source == null) { + throw new IllegalArgumentException("sources must not contain null elements"); + } + } + } + + /** + * Builds the aggregate {@link CompletionException} for {@code first} + * when every source fails. The first failure is the cause; remaining + * failures are attached as suppressed exceptions. + */ + private static CompletionException aggregateFirstFailures(int sourceCount, List errors) { + CompletionException aggregate = new CompletionException( + "All " + sourceCount + " tasks failed", errors.get(0)); + for (int i = 1; i < errors.size(); i++) { + aggregate.addSuppressed(errors.get(i)); + } + return aggregate; + } + + @SuppressWarnings("unchecked") + private static List getJoinedResults(CompletableFuture[] futures) { + List results = new ArrayList<>(futures.length); + for (CompletableFuture future : futures) { + results.add((T) future.join()); + } + return results; + } + + private static List> getAwaitResults(CompletableFuture[] futures) { + List> results = new ArrayList<>(futures.length); + for (CompletableFuture future : futures) { + try { + results.add(AwaitResult.success(future.join())); + } catch (CompletionException e) { + results.add(AwaitResult.failure(AsyncSupport.unwrap(e))); + } catch (CancellationException e) { + results.add(AwaitResult.failure(e)); + } + } + return results; + } + + private static CompletionException asCompletionException(Throwable error) { + return error instanceof CompletionException ce ? ce : new CompletionException(error); + } +} diff --git a/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java b/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java index 044b948d0fd..fa8bdf6803f 100644 --- a/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java +++ b/src/main/java/org/apache/groovy/runtime/async/GroovyPromise.java @@ -18,7 +18,6 @@ */ package org.apache.groovy.runtime.async; -// AsyncContext support reserved for future enhancement import groovy.concurrent.Awaitable; import java.util.Objects; @@ -179,9 +178,7 @@ public boolean isCompletedExceptionally() { * {@inheritDoc} *

* Returns a new {@code GroovyPromise} whose result is obtained by applying - * the given function to this promise's result. The current - * {@code AsyncContext} snapshot is captured when the continuation is - * registered and restored when it executes. + * the given function to this promise's result when it completes. */ @Override public Awaitable then(Function fn) { @@ -193,8 +190,6 @@ public Awaitable then(Function fn) { *

* Returns a new {@code GroovyPromise} that is the result of composing this * promise with the async function, enabling flat-mapping of awaitables. - * The current {@code AsyncContext} snapshot is captured when the - * continuation is registered and restored when it executes. */ @Override public Awaitable thenCompose(Function> fn) { @@ -208,8 +203,6 @@ public Awaitable thenCompose(Function> * Returns a new {@code GroovyPromise} that handles exceptions thrown by this promise. * The throwable passed to the handler is deeply unwrapped to strip JDK * wrapper layers ({@code CompletionException}, {@code ExecutionException}). - * The handler runs with the {@code AsyncContext} snapshot that was active - * when the recovery continuation was registered. */ @Override public Awaitable exceptionally(Function fn) { diff --git a/src/main/java/org/apache/groovy/runtime/async/package-info.java b/src/main/java/org/apache/groovy/runtime/async/package-info.java index f40eece142e..5b03865bd0e 100644 --- a/src/main/java/org/apache/groovy/runtime/async/package-info.java +++ b/src/main/java/org/apache/groovy/runtime/async/package-info.java @@ -18,6 +18,24 @@ */ /** - * Runtime support for async/await expressions. Provides continuation-passing style transformation and execution framework. + * Runtime support for Groovy's {@code async}/{@code await}/{@code defer} + * language features. + *

+ * Layout: + *

    + *
  • {@link org.apache.groovy.runtime.async.AsyncSupport} — public entry + * point used by compiler-generated code and by + * {@link groovy.concurrent.Awaitable}
    • + *
  • {@code AsyncExecutors} — executor/scheduler configuration
    • + *
  • {@code AwaitCombinators} — {@code all}/{@code any}/{@code first}/ + * {@code allSettled} joining algorithms
    • + *
  • {@link org.apache.groovy.runtime.async.GroovyPromise} — default + * {@link groovy.concurrent.Awaitable} implementation
    • + *
  • {@link org.apache.groovy.runtime.async.GeneratorBridge} — + * producer/consumer bridge for {@code yield return}
    • + *
  • {@link org.apache.groovy.runtime.async.DefaultAsyncScope}, + * {@link org.apache.groovy.runtime.async.DefaultAsyncChannel} — + * structured concurrency and CSP channel implementations
    • + *
*/ package org.apache.groovy.runtime.async; diff --git a/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java b/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java index f4275e62461..41f96d184db 100644 --- a/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java +++ b/src/main/java/org/codehaus/groovy/transform/AsyncTransformHelper.java @@ -22,20 +22,22 @@ import org.apache.groovy.runtime.async.AsyncSupport; import org.codehaus.groovy.ast.ClassHelper; import org.codehaus.groovy.ast.ClassNode; -import org.codehaus.groovy.ast.CodeVisitorSupport; import org.codehaus.groovy.ast.Parameter; import org.codehaus.groovy.ast.expr.ArgumentListExpression; import org.codehaus.groovy.ast.expr.ClosureExpression; import org.codehaus.groovy.ast.expr.Expression; import org.codehaus.groovy.ast.expr.StaticMethodCallExpression; import org.codehaus.groovy.ast.query.AstQuery; +import org.codehaus.groovy.ast.stmt.ForStatement; import org.codehaus.groovy.ast.stmt.Statement; +import org.codehaus.groovy.ast.stmt.TryCatchStatement; + +import java.util.concurrent.atomic.AtomicLong; import static org.codehaus.groovy.ast.tools.GeneralUtils.args; import static org.codehaus.groovy.ast.tools.GeneralUtils.block; import static org.codehaus.groovy.ast.tools.GeneralUtils.callX; import static org.codehaus.groovy.ast.tools.GeneralUtils.castX; -import static org.codehaus.groovy.ast.tools.GeneralUtils.classX; import static org.codehaus.groovy.ast.tools.GeneralUtils.declS; import static org.codehaus.groovy.ast.tools.GeneralUtils.stmt; import static org.codehaus.groovy.ast.tools.GeneralUtils.tryCatchS; @@ -44,7 +46,9 @@ /** * Shared AST utilities for the {@code async}/{@code await}/{@code defer} language features. *

- * Centralises AST node construction for the parser ({@code AstBuilder}). + * Centralises AST node construction and higher-level rewrites used by the + * parser ({@code AstBuilder}) so feature logic does not sprawl across the + * visitor. * * @since 6.0.0 */ @@ -65,6 +69,9 @@ public final class AsyncTransformHelper { private static final String DEFER_METHOD = "defer"; private static final String EXECUTE_DEFER_SCOPE_METHOD = "executeDeferScope"; + /** Monotonic suffix for synthetic {@code for await} source variables. */ + private static final AtomicLong FOR_AWAIT_SEQ = new AtomicLong(); + private AsyncTransformHelper() { } private static Expression ensureArgs(Expression expr) { @@ -74,17 +81,25 @@ private static Expression ensureArgs(Expression expr) { /** * Builds {@code AsyncSupport.await(arg)} or for multi-arg: * {@code AsyncSupport.await(Awaitable.all(arg1, arg2, ...))}. + *

+ * Single-arg form casts to {@link Object} so static/dynamic overload + * selection targets {@link AsyncSupport#await(Object)}, avoiding + * ambiguity when the argument implements multiple async interfaces + * (e.g. {@link java.util.concurrent.CompletableFuture} implements both + * {@link java.util.concurrent.CompletionStage} and + * {@link java.util.concurrent.Future}). */ public static Expression buildAwaitCall(Expression arg) { if (arg instanceof ArgumentListExpression args && args.getExpressions().size() > 1) { - Expression allCall = callX(classX(AWAITABLE_TYPE), "all", args); + Expression allCall = callX(AWAITABLE_TYPE, "all", args); return callX(ASYNC_SUPPORT_TYPE, AWAIT_METHOD, new ArgumentListExpression(allCall)); } - // Cast to Object to force dynamic dispatch to the await(Object) overload, - // avoiding ambiguity when the argument implements multiple async interfaces - // (e.g., CompletableFuture implements both CompletionStage and Future) + Expression source = arg; + if (arg instanceof ArgumentListExpression args && !args.getExpressions().isEmpty()) { + source = args.getExpression(0); + } return callX(ASYNC_SUPPORT_TYPE, AWAIT_METHOD, - new ArgumentListExpression(castX(ClassHelper.OBJECT_TYPE, arg))); + new ArgumentListExpression(castX(ClassHelper.OBJECT_TYPE, source))); } /** @@ -143,34 +158,12 @@ public static Parameter createGenParam() { public static boolean containsYieldReturn(Statement stmt) { return AstQuery.from(stmt) .descendants(StaticMethodCallExpression.class) - .notInto(ClosureExpression.class) // don't descend into nested closures + .notInto(ClosureExpression.class) .where(call -> YIELD_RETURN_METHOD.equals(call.getMethod()) && AsyncSupport.class.getName().equals(call.getOwnerType().getName())) .any(); } - /** - * Rewrites {@code yieldReturn(expr)} calls to {@code yieldReturn($__asyncGen__, expr)} - * by injecting the generator parameter reference. - */ - public static void injectGenParamIntoYieldReturnCalls(Statement stmt, Parameter genParam) { - stmt.visit(new CodeVisitorSupport() { - @Override - public void visitStaticMethodCallExpression(StaticMethodCallExpression call) { - if (YIELD_RETURN_METHOD.equals(call.getMethod()) - && AsyncSupport.class.getName().equals(call.getOwnerType().getName())) { - // Already built with gen param by buildYieldReturnCall — no action needed - // This method is a hook point for future transformations - } - super.visitStaticMethodCallExpression(call); - } - @Override - public void visitClosureExpression(ClosureExpression expression) { - // Don't descend into nested closures - } - }); - } - /** * Returns {@code true} if the statement tree contains a {@code defer} call, * without descending into nested closures. @@ -178,7 +171,7 @@ public void visitClosureExpression(ClosureExpression expression) { public static boolean containsDefer(Statement stmt) { return AstQuery.from(stmt) .descendants(StaticMethodCallExpression.class) - .notInto(ClosureExpression.class) // don't descend into nested closures + .notInto(ClosureExpression.class) .where(call -> DEFER_METHOD.equals(call.getMethod()) && AsyncSupport.class.getName().equals(call.getOwnerType().getName())) .any(); @@ -193,14 +186,80 @@ public static boolean containsDefer(Statement stmt) { * */ public static Statement wrapWithDeferScope(Statement body) { - // var $__deferScope__ = AsyncSupport.createDeferScope() Statement declStmt = declS(varX(DEFER_SCOPE_VAR), callX(ASYNC_SUPPORT_TYPE, CREATE_DEFER_SCOPE_METHOD)); - // try { body } finally { AsyncSupport.executeDeferScope($__deferScope__) } Statement finallyStmt = stmt(callX(ASYNC_SUPPORT_TYPE, EXECUTE_DEFER_SCOPE_METHOD, args(varX(DEFER_SCOPE_VAR)))); return block(declStmt, tryCatchS(body, finallyStmt)); } + + /** + * Rewrites a {@code for await} loop into a block that materialises the + * source iterable, runs the loop, and closes the source in a finally block: + * 

+     * var $__forAwaitSource_N = AsyncSupport.toIterable(collection)
+     * try {
+     *     for (... in $__forAwaitSource_N) { body }
+     * } finally {
+     *     AsyncSupport.closeIterable($__forAwaitSource_N)
+     * }
+     *
+ * + * @param forStatement the for statement whose collection is the await source + * @return the rewritten statement block + */ + public static Statement wrapForAwaitLoop(ForStatement forStatement) { + Expression original = forStatement.getCollectionExpression(); + String tempVar = "$__forAwaitSource_" + FOR_AWAIT_SEQ.incrementAndGet(); + Expression toIterableCall = buildToIterableCall(original); + + Statement declStmt = declS(varX(tempVar), toIterableCall); + forStatement.setCollectionExpression(varX(tempVar)); + + Statement finallyStmt = stmt(buildCloseIterableCall(varX(tempVar))); + TryCatchStatement tryCatch = new TryCatchStatement(forStatement, finallyStmt); + return block(declStmt, tryCatch); + } + + /** + * Applies defer-scope wrapping and generator parameter injection to an + * {@code async { ... }} closure, then builds the runtime call + * ({@code async} or {@code asyncGenerator}). + * + * @param closure the parsed async closure expression + * @return the desugared runtime call expression + */ + public static Expression transformAsyncClosure(ClosureExpression closure) { + boolean hasYieldReturn = containsYieldReturn(closure.getCode()); + boolean hasDefer = containsDefer(closure.getCode()); + + if (hasDefer) { + Statement wrappedBody = wrapWithDeferScope(closure.getCode()); + ClosureExpression newClosure = new ClosureExpression(closure.getParameters(), wrappedBody); + newClosure.setVariableScope(closure.getVariableScope()); + newClosure.setSourcePosition(closure); + closure = newClosure; + } + + if (hasYieldReturn) { + Parameter genParam = createGenParam(); + Parameter[] existingParams = closure.getParameters(); + boolean hasUserParams = existingParams != null && existingParams.length > 0; + Parameter[] newParams; + if (hasUserParams) { + newParams = new Parameter[existingParams.length + 1]; + newParams[0] = genParam; + System.arraycopy(existingParams, 0, newParams, 1, existingParams.length); + } else { + newParams = new Parameter[]{genParam}; + } + ClosureExpression genClosure = new ClosureExpression(newParams, closure.getCode()); + genClosure.setVariableScope(closure.getVariableScope()); + genClosure.setSourcePosition(closure); + return buildAsyncGeneratorCall(new ArgumentListExpression(genClosure)); + } + return buildAsyncCall(new ArgumentListExpression(closure)); + } } diff --git a/src/spec/doc/core-async-await.adoc b/src/spec/doc/core-async-await.adoc index 858a52bcda7..dfff725c311 100644 --- a/src/spec/doc/core-async-await.adoc +++ b/src/spec/doc/core-async-await.adoc @@ -359,7 +359,9 @@ import org.apache.groovy.runtime.async.AsyncSupport import java.util.concurrent.Executors AsyncSupport.setExecutor(Executors.newFixedThreadPool(4)) -AsyncSupport.resetExecutor() // restore default +AsyncSupport.setExecutor(null) // restore platform default +// or equivalently: +AsyncSupport.resetExecutor() ---- [[async-jdk-integration]] diff --git a/src/test/groovy/groovy/AsyncAwaitTest.groovy b/src/test/groovy/groovy/AsyncAwaitTest.groovy index ffc5826a348..92085273c87 100644 --- a/src/test/groovy/groovy/AsyncAwaitTest.groovy +++ b/src/test/groovy/groovy/AsyncAwaitTest.groovy @@ -1116,11 +1116,51 @@ final class AsyncAwaitTest { await Awaitable.first(f1, f2) assert false : 'should have thrown' } catch (Exception e) { + // await exception transparency peels the aggregate CompletionException assert e != null } ''' } + @Test + void testSetExecutorNullResetsToDefault() { + assertScript ''' + import groovy.concurrent.Awaitable + import org.apache.groovy.runtime.async.AsyncSupport + import java.util.concurrent.Executors + + def original = Awaitable.getExecutor() + def custom = Executors.newSingleThreadExecutor() + try { + Awaitable.setExecutor(custom) + assert Awaitable.getExecutor().is(custom) + Awaitable.setExecutor(null) + assert Awaitable.getExecutor() != null + assert !Awaitable.getExecutor().is(custom) + // AsyncSupport path is equivalent + AsyncSupport.setExecutor(custom) + AsyncSupport.setExecutor(null) + assert AsyncSupport.getExecutor() != null + } finally { + Awaitable.setExecutor(original) + custom.shutdown() + } + ''' + } + + @Test + void testAwaitObjectFallbackHandlesNullAndAwaitable() { + assertScript ''' + import org.apache.groovy.runtime.async.AsyncSupport + import groovy.concurrent.Awaitable + + assert AsyncSupport.await((Object) null) == null + assert AsyncSupport.await((Object) Awaitable.of(42)) == 42 + def task = async { 'x' } + assert AsyncSupport.await((Object) task) == 'x' + ''' + } + @Test void testAllSettledWithCancelledTask() { assertScript ''' diff --git a/src/test/groovy/groovy/concurrent/AwaitResultTest.groovy b/src/test/groovy/groovy/concurrent/AwaitResultTest.groovy new file mode 100644 index 00000000000..72da8c98e61 --- /dev/null +++ b/src/test/groovy/groovy/concurrent/AwaitResultTest.groovy @@ -0,0 +1,118 @@ +/* + * 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 org.junit.jupiter.api.Test + +import static groovy.test.GroovyAssert.shouldFail + +/** + * Unit tests for {@link AwaitResult} value-object behaviour. + */ +final class AwaitResultTest { + + @Test + void testSuccessHoldsValueIncludingNull() { + def ok = AwaitResult.success('hello') + assert ok.success + assert !ok.failure + assert ok.value == 'hello' + assert ok.toString() == 'AwaitResult.Success[hello]' + + def nil = AwaitResult.success(null) + assert nil.success + assert nil.value == null + } + + @Test + void testFailureHoldsError() { + def err = new RuntimeException('boom') + def fail = AwaitResult.failure(err) + assert fail.failure + assert !fail.success + assert fail.error.is(err) + assert fail.toString().contains('boom') + } + + @Test + void testFailureRejectsNullError() { + shouldFail(NullPointerException) { + AwaitResult.failure(null) + } + } + + @Test + void testGetValueOnFailureThrows() { + def fail = AwaitResult.failure(new Exception('x')) + def e = shouldFail(IllegalStateException) { + fail.value + } + assert e.message.contains('failed') + } + + @Test + void testGetErrorOnSuccessThrows() { + def ok = AwaitResult.success(1) + def e = shouldFail(IllegalStateException) { + ok.error + } + assert e.message.contains('successful') + } + + @Test + void testGetOrElse() { + assert AwaitResult.success(7).getOrElse { -1 } == 7 + assert AwaitResult.failure(new Exception('e')).getOrElse { it.message.length() } == 1 + } + + @Test + void testMapOnSuccessAndFailure() { + assert AwaitResult.success('ab').map { it.length() }.value == 2 + def fail = AwaitResult.failure(new Exception('keep')) + def mapped = fail.map { it.toUpperCase() } + assert mapped.failure + assert mapped.error.message == 'keep' + } + + @Test + void testMapRejectsNullFunction() { + shouldFail(NullPointerException) { + AwaitResult.success(1).map(null) + } + } + + @Test + void testEqualsAndHashCode() { + def a = AwaitResult.success('x') + def b = AwaitResult.success('x') + def c = AwaitResult.success('y') + def err = new RuntimeException('e') + def f1 = AwaitResult.failure(err) + def f2 = AwaitResult.failure(err) + + assert a == b + assert a.hashCode() == b.hashCode() + assert a != c + assert a != f1 + assert f1 == f2 + assert f1.hashCode() == f2.hashCode() + assert a != null + assert a != 'x' + } +} diff --git a/src/test/groovy/org/apache/groovy/runtime/async/AsyncSupportInternalsTest.groovy b/src/test/groovy/org/apache/groovy/runtime/async/AsyncSupportInternalsTest.groovy new file mode 100644 index 00000000000..ce237007e59 --- /dev/null +++ b/src/test/groovy/org/apache/groovy/runtime/async/AsyncSupportInternalsTest.groovy @@ -0,0 +1,259 @@ +/* + * 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 org.apache.groovy.runtime.async + +import groovy.concurrent.Awaitable +import org.junit.jupiter.api.Test + +import java.util.concurrent.CompletableFuture +import java.util.concurrent.CompletionException +import java.util.concurrent.CompletionStage +import java.util.concurrent.ExecutionException +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.TimeoutException + +import static groovy.test.GroovyAssert.shouldFail + +/** + * Focused unit coverage for the async runtime refactor + * ({@link AsyncSupport}, {@link AsyncExecutors}, {@link AwaitCombinators}). + */ +final class AsyncSupportInternalsTest { + + @Test + void testGoIsAliasOfAsync() { + def viaGo = AsyncSupport.go { 11 } + def viaAsync = AsyncSupport.async { 22 } + assert AsyncSupport.await(viaGo) == 11 + assert AsyncSupport.await(viaAsync) == 22 + } + + @Test + void testAwaitObjectOnCompletableFutureImplementsMultipleInterfaces() { + // CompletableFuture is CompletionStage and Future; await(Object) is unambiguous + CompletableFuture cf = CompletableFuture.completedFuture('cf') + assert AsyncSupport.await((Object) cf) == 'cf' + assert AsyncSupport.await((CompletionStage) cf) == 'cf' + assert AsyncSupport.await(cf) == 'cf' + } + + @Test + void testAwaitUnwrapsNestedWrappers() { + def cause = new IllegalStateException('root') + def cf = new CompletableFuture() + cf.completeExceptionally(new CompletionException(new ExecutionException(cause))) + def thrown = shouldFail(IllegalStateException) { + AsyncSupport.await(cf) + } + assert thrown.is(cause) + } + + @Test + void testAllAsyncEmptyAndSingle() { + assert AsyncSupport.await(AsyncSupport.allAsync()) == [] + assert AsyncSupport.await(AsyncSupport.allAsync(Awaitable.of(1))) == [1] + assert AsyncSupport.all() == [] + assert AsyncSupport.all(Awaitable.of('a'), Awaitable.of('b')) == ['a', 'b'] + } + + @Test + void testAllPropagatesOriginalFailureNotWrapper() { + def fail = Awaitable.failed(new RuntimeException('all-x')) + def ok = Awaitable.of(1) + def thrown = shouldFail(RuntimeException) { + AsyncSupport.all(ok, fail) + } + assert thrown.message == 'all-x' + } + + @Test + void testFirstAllFailThrowsUnwrappedCause() { + def f1 = Awaitable.failed(new RuntimeException('e1')) + def f2 = Awaitable.failed(new RuntimeException('e2')) + // Aggregate is a CompletionException; await transparency peels to the cause + def thrown = shouldFail(RuntimeException) { + AsyncSupport.await(AsyncSupport.firstAsync(f1, f2)) + } + assert thrown.message == 'e1' + } + + @Test + void testFirstRejectsNullArrayAndNullElementAndEmpty() { + shouldFail(IllegalArgumentException) { + AsyncSupport.firstAsync((Object[]) null) + } + shouldFail(IllegalArgumentException) { + AsyncSupport.firstAsync() + } + shouldFail(IllegalArgumentException) { + AsyncSupport.firstAsync(Awaitable.of(1), null) + } + } + + @Test + void testAnyAsyncPicksCompletedValue() { + def result = AsyncSupport.await(AsyncSupport.anyAsync( + Awaitable.of('winner'), + asyncSleep(500, 'loser'))) + assert result == 'winner' + } + + @Test + void testAllSettledMixed() { + def results = AsyncSupport.allSettled( + Awaitable.of('ok'), + Awaitable.failed(new RuntimeException('bad'))) + assert results.size() == 2 + assert results[0].success && results[0].value == 'ok' + assert results[1].failure && results[1].error.message == 'bad' + } + + @Test + void testSetExecutorNullAndReset() { + def original = AsyncSupport.getExecutor() + def custom = Executors.newSingleThreadExecutor() + try { + AsyncSupport.setExecutor(custom) + assert AsyncSupport.getExecutor().is(custom) + AsyncSupport.setExecutor(null) + assert AsyncSupport.getExecutor() != null + assert !AsyncSupport.getExecutor().is(custom) + AsyncSupport.setExecutor(custom) + AsyncSupport.resetExecutor() + assert !AsyncSupport.getExecutor().is(custom) + } finally { + AsyncSupport.setExecutor(original) + custom.shutdownNow() + } + } + + @Test + void testDelayAndTimeouts() { + assert AsyncSupport.await(AsyncSupport.delay(1)) == null + assert AsyncSupport.await(AsyncSupport.delay(1, TimeUnit.MILLISECONDS)) == null + + def slow = asyncSleep(5_000, 'late') + def timedOut = shouldFail(TimeoutException) { + AsyncSupport.await(AsyncSupport.orTimeoutMillis(slow, 20)) + } + assert timedOut.message.contains('Timed out') + + def fallback = AsyncSupport.await( + AsyncSupport.completeOnTimeoutMillis(asyncSleep(5_000, 'late'), 'fb', 20)) + assert fallback == 'fb' + } + + @Test + void testToIterableAndCloseIterable() { + assert AsyncSupport.toIterable(null).iterator().toList() == [] + assert AsyncSupport.toIterable([1, 2]).iterator().toList() == [1, 2] + assert AsyncSupport.toIterable((Object[]) ['a', 'b']).iterator().toList() == ['a', 'b'] + def it = [9].iterator() + assert AsyncSupport.toIterable(it).iterator().is(it) + + def closed = new boolean[1] + def closeable = new Closeable() { + @Override void close() { closed[0] = true } + } + AsyncSupport.closeIterable(closeable) + assert closed[0] + + def autoClosed = new boolean[1] + def auto = new AutoCloseable() { + @Override void close() { autoClosed[0] = true } + } + AsyncSupport.closeIterable(auto) + assert autoClosed[0] + + // non-closeable is a no-op + AsyncSupport.closeIterable('plain') + } + + @Test + void testCloseIterableSwallowsCloseException() { + def bad = new Closeable() { + @Override void close() { throw new IOException('ignore-me') } + } + AsyncSupport.closeIterable(bad) // must not throw + } + + @Test + void testDeferScopeLIFOAndNullGuards() { + def log = [] + def scope = AsyncSupport.createDeferScope() + AsyncSupport.defer(scope, { log << 'a'; null }) + AsyncSupport.defer(scope, { log << 'b'; null }) + AsyncSupport.executeDeferScope(scope) + assert log == ['b', 'a'] + + shouldFail(IllegalStateException) { + AsyncSupport.defer(null, { null }) + } + shouldFail(IllegalArgumentException) { + AsyncSupport.defer(AsyncSupport.createDeferScope(), null) + } + AsyncSupport.executeDeferScope(null) // no-op + AsyncSupport.executeDeferScope(AsyncSupport.createDeferScope()) // empty no-op + } + + @Test + void testYieldReturnRequiresGeneratorBridge() { + shouldFail(IllegalStateException) { + AsyncSupport.yieldReturn('not-a-bridge', 1) + } + } + + @Test + void testAsyncGeneratorProducesValues() { + def values = AsyncSupport.asyncGenerator { bridge -> + AsyncSupport.yieldReturn(bridge, 1) + AsyncSupport.yieldReturn(bridge, 2) + }.iterator().toList() + assert values == [1, 2] + } + + @Test + void testWrapForFutureAndUnwrap() { + def ce = new CompletionException(new RuntimeException('x')) + assert AsyncSupport.wrapForFuture(ce).is(ce) + def wrapped = AsyncSupport.wrapForFuture(new RuntimeException('y')) + assert wrapped instanceof CompletionException + assert AsyncSupport.unwrap(wrapped).message == 'y' + } + + @Test + void testIsVirtualThreadsAvailableIsConsistent() { + boolean available = AsyncSupport.isVirtualThreadsAvailable() + assert available == Awaitable.isVirtualThreadsAvailable() + } + + @Test + void testSchedulerIsDaemon() { + def scheduler = AsyncSupport.getScheduler() + assert scheduler != null + } + + private static Awaitable asyncSleep(long millis, Object value) { + AsyncSupport.async { + Thread.sleep(millis) + value + } + } +} diff --git a/src/test/groovy/org/codehaus/groovy/transform/AsyncTransformHelperTest.groovy b/src/test/groovy/org/codehaus/groovy/transform/AsyncTransformHelperTest.groovy new file mode 100644 index 00000000000..129591aede2 --- /dev/null +++ b/src/test/groovy/org/codehaus/groovy/transform/AsyncTransformHelperTest.groovy @@ -0,0 +1,184 @@ +/* + * 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 org.codehaus.groovy.transform + +import org.codehaus.groovy.ast.ClassHelper +import org.codehaus.groovy.ast.Parameter +import org.codehaus.groovy.ast.expr.ArgumentListExpression +import org.codehaus.groovy.ast.expr.CastExpression +import org.codehaus.groovy.ast.expr.ClosureExpression +import org.codehaus.groovy.ast.expr.ConstantExpression +import org.codehaus.groovy.ast.expr.MethodCallExpression +import org.codehaus.groovy.ast.expr.StaticMethodCallExpression +import org.codehaus.groovy.ast.expr.VariableExpression +import org.codehaus.groovy.ast.stmt.BlockStatement +import org.codehaus.groovy.ast.stmt.EmptyStatement +import org.codehaus.groovy.ast.stmt.ExpressionStatement +import org.codehaus.groovy.ast.stmt.ForStatement +import org.codehaus.groovy.ast.tools.GeneralUtils +import org.junit.jupiter.api.Test + +/** + * Unit tests for {@link AsyncTransformHelper} AST construction helpers. + */ +final class AsyncTransformHelperTest { + + @Test + void testBuildAwaitCallSingleArgUsesAwaitWithObjectCast() { + def expr = AsyncTransformHelper.buildAwaitCall(new ConstantExpression(1)) + assert methodName(expr) == 'await' + assert ownerName(expr).endsWith('AsyncSupport') + def arg0 = ((StaticMethodCallExpression) expr).arguments.expressions[0] + assert arg0 instanceof CastExpression + assert arg0.type == ClassHelper.OBJECT_TYPE + } + + @Test + void testBuildAwaitCallMultiArgUsesAwaitableAllThenAwait() { + def args = new ArgumentListExpression( + new ConstantExpression(1), + new ConstantExpression(2)) + def expr = AsyncTransformHelper.buildAwaitCall(args) + assert methodName(expr) == 'await' + def arg0 = ((StaticMethodCallExpression) expr).arguments.expressions[0] + assert methodName(arg0) == 'all' + assert ownerName(arg0).endsWith('Awaitable') + } + + @Test + void testContainsYieldReturnAndDeferDoNotDescendIntoNestedClosures() { + def yieldCall = AsyncTransformHelper.buildYieldReturnCall(new ConstantExpression(1)) + def deferCall = AsyncTransformHelper.buildDeferCall( + new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE)) + + def outerWithYield = new BlockStatement() + outerWithYield.addStatement(new ExpressionStatement(yieldCall)) + assert AsyncTransformHelper.containsYieldReturn(outerWithYield) + assert !AsyncTransformHelper.containsDefer(outerWithYield) + + def outerWithDefer = new BlockStatement() + outerWithDefer.addStatement(new ExpressionStatement(deferCall)) + assert AsyncTransformHelper.containsDefer(outerWithDefer) + assert !AsyncTransformHelper.containsYieldReturn(outerWithDefer) + + // Nested inside a child closure — must be ignored + def nested = new ClosureExpression(Parameter.EMPTY_ARRAY, outerWithYield) + def outerOnlyNested = new BlockStatement() + outerOnlyNested.addStatement(new ExpressionStatement(nested)) + assert !AsyncTransformHelper.containsYieldReturn(outerOnlyNested) + } + + @Test + void testWrapWithDeferScopeStructure() { + def body = EmptyStatement.INSTANCE + def wrapped = AsyncTransformHelper.wrapWithDeferScope(body) + assert wrapped instanceof BlockStatement + assert wrapped.statements.size() == 2 + } + + @Test + void testWrapForAwaitLoopUsesUniqueSyntheticNames() { + def collection = new ConstantExpression([1, 2, 3]) + def loop1 = new ForStatement( + new Parameter(org.codehaus.groovy.ast.ClassHelper.OBJECT_TYPE, 'item'), + collection, + EmptyStatement.INSTANCE) + def loop2 = new ForStatement( + new Parameter(org.codehaus.groovy.ast.ClassHelper.OBJECT_TYPE, 'item'), + collection, + EmptyStatement.INSTANCE) + + def block1 = AsyncTransformHelper.wrapForAwaitLoop(loop1) as BlockStatement + def block2 = AsyncTransformHelper.wrapForAwaitLoop(loop2) as BlockStatement + + def name1 = extractDeclaredName(block1) + def name2 = extractDeclaredName(block2) + assert name1.startsWith('$__forAwaitSource_') + assert name2.startsWith('$__forAwaitSource_') + assert name1 != name2 + } + + @Test + void testTransformAsyncClosurePlainAndGenerator() { + def plain = new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE) + def plainCall = AsyncTransformHelper.transformAsyncClosure(plain) + assert methodName(plainCall) == 'async' + + def yieldCall = AsyncTransformHelper.buildYieldReturnCall(new ConstantExpression(1)) + def yieldBody = new BlockStatement() + yieldBody.addStatement(new ExpressionStatement(yieldCall)) + def genClosure = new ClosureExpression(Parameter.EMPTY_ARRAY, yieldBody) + def genCall = AsyncTransformHelper.transformAsyncClosure(genClosure) + assert methodName(genCall) == 'asyncGenerator' + } + + @Test + void testTransformAsyncClosureWithDefer() { + def deferCall = AsyncTransformHelper.buildDeferCall( + new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE)) + def body = new BlockStatement() + body.addStatement(new ExpressionStatement(deferCall)) + def closure = new ClosureExpression(Parameter.EMPTY_ARRAY, body) + def call = AsyncTransformHelper.transformAsyncClosure(closure) + assert methodName(call) == 'async' + } + + @Test + void testCreateGenParamAndBuilders() { + def p = AsyncTransformHelper.createGenParam() + assert p.name == '$__asyncGen__' + + assert methodName(AsyncTransformHelper.buildAsyncCall( + new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE))) == 'async' + assert methodName(AsyncTransformHelper.buildAsyncGeneratorCall( + new ClosureExpression(Parameter.EMPTY_ARRAY, EmptyStatement.INSTANCE))) == 'asyncGenerator' + assert methodName(AsyncTransformHelper.buildToIterableCall(new ConstantExpression([]))) == 'toIterable' + assert methodName(AsyncTransformHelper.buildCloseIterableCall(GeneralUtils.varX('x'))) == 'closeIterable' + } + + private static String methodName(Object expr) { + if (expr instanceof StaticMethodCallExpression) { + return expr.method + } + if (expr instanceof MethodCallExpression) { + return expr.methodAsString + } + throw new AssertionError("unexpected expression: ${expr?.getClass()?.name}") + } + + private static String ownerName(Object expr) { + if (expr instanceof StaticMethodCallExpression) { + return expr.ownerType.name + } + if (expr instanceof MethodCallExpression) { + return expr.objectExpression.type.name + } + throw new AssertionError("unexpected expression: ${expr?.getClass()?.name}") + } + + private static String extractDeclaredName(BlockStatement block) { + def declStmt = block.statements[0] as ExpressionStatement + def decl = declStmt.expression + def left = decl.leftExpression + if (left instanceof VariableExpression) { + return left.name + } + return decl.variableExpression.name + } +}