Merge pull request #442 from KyleAure/439-sched-async-tests

Add scheduled async tests

api/src/main/java/jakarta/enterprise/concurrent/Asynchronous.java

@@ -199,7 +199,7 @@
      * attempts to start the next execution of the method at 8:03 AM.</p>
      *
      * <p>Scheduled asynchronous methods are treated similar to other scheduled
-     * tasks in that they are not subject to {@code max-async} constaints of
+     * tasks in that they are not subject to {@code max-async} constraints of
      * {@code managed-scheduled-executor-definition} and
      * {@code managed-executor-definition} and the corresponding
      * {@link ManagedScheduledExecutorDefinition#maxAsync()} and

tck/src/main/java/ee/jakarta/tck/concurrent/spec/ManagedExecutorService/resourcedef/AppBean.java

@@ -15,6 +15,7 @@
  */
 package ee.jakarta.tck.concurrent.spec.ManagedExecutorService.resourcedef;
 
+import java.time.Duration;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -24,15 +25,23 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Logger;
 
 import ee.jakarta.tck.concurrent.common.context.IntContext;
 import ee.jakarta.tck.concurrent.common.context.StringContext;
 import jakarta.enterprise.concurrent.Asynchronous;
+import jakarta.enterprise.concurrent.Schedule;
 import jakarta.enterprise.context.ApplicationScoped;
 
 @ApplicationScoped
 public class AppBean {
+
+    private static final Logger log = Logger.getLogger(AppBean.class.getCanonicalName());
+
     private static final long MAX_WAIT_SECONDS = TimeUnit.MINUTES.toSeconds(2);
+
+    // Asynchronous Tests
 
     @Asynchronous(executor = "java:module/concurrent/ExecutorB")
     public CompletionStage<String> addStringContextAndWait(final BlockingQueue<String> queue, final CountDownLatch blocker) {
@@ -70,4 +79,188 @@ public CompletableFuture<Integer> waitAndGetIntContext(final Semaphore started,
         }
         return future;
     }
+
+    // Scheduled Asynchronous Tests
+    public static enum RETURN {
+            NULL, // Never completes
+            COMPLETE_EXCEPTIONALLY, // Completes with exception
+            COMPLETE_RESULT, // Completes with result
+            INCOMPLETE, // Returns future that is not complete
+            THROW_EXCEPTION; // Method throws exception
+
+            private String message = "";
+
+            public RETURN withMessage(final String m) {
+                this.message = m;
+                return this;
+            }
+
+            public String getMessage() {
+                return message;
+            }
+    }
+
+    /**
+     * A scheduled async method that runs every 5 seconds
+     *
+     * @param runs - how many times to run before returning
+     * @param type - how this method should return (successfully / incomplete / exceptionally)
+     * @param counter - The counter provided from the caller to compare against
+     *
+     * @return A result or exception
+     */
+    @Asynchronous(runAt = @Schedule(cron = "*/5 * * * * *"))
+    public CompletableFuture<Integer> scheduledEvery5seconds(final int runs, final RETURN type, final AtomicInteger counter) {
+        int count = counter.incrementAndGet();
+
+        log.info("Executing scheduledEvery5seconds method " + count + "/" + runs + " (Returning: " + type.toString() + ")");
+        log.info("  Thread: " + Thread.currentThread().toString());
+
+        if (runs != count) {
+            return null; // Continue onto next scheduled execution
+        }
+
+        CompletableFuture<Integer> future = Asynchronous.Result.getFuture();
+
+        switch (type) {
+        case NULL: //Never stop executions
+            return null;
+        case COMPLETE_EXCEPTIONALLY:
+            future.completeExceptionally(new Exception(type.getMessage()));
+            break;
+        case COMPLETE_RESULT:
+            future.complete(count);
+            break;
+        case INCOMPLETE:
+            break; //never complete future
+        case THROW_EXCEPTION:
+            throw new RuntimeException(type.getMessage());
+        default:
+            break;
+        }
+
+        return future;
+    }
+
+    @Asynchronous(runAt = @Schedule(cron = "*/3 * * * * *"))
+    public void scheduledEvery3SecondsVoidReturn(final int runs, final RETURN type, final AtomicInteger counter) {
+        int count = counter.incrementAndGet();
+
+        log.info("Executing scheduledEvery3SecondsVoidReturn method " + count + "/" + runs + " (Returning: " + type.toString() + ")");
+        log.info("  Thread: " + Thread.currentThread().toString());
+
+        if (runs != count) {
+            return; // Continue onto next scheduled execution
+        }
+
+        CompletableFuture<Void> future = Asynchronous.Result.getFuture();
+
+        switch (type) {
+        case COMPLETE_EXCEPTIONALLY:
+            future.completeExceptionally(new Exception(type.getMessage()));
+            break;
+        case COMPLETE_RESULT:
+            future.complete(null);
+            break;
+        case THROW_EXCEPTION:
+            throw new RuntimeException(type.getMessage());
+        default:
+            break;
+        }
+    }
+
+    /**
+     * A scheduled async method that runs every 3 seconds, but takes 5 seconds to complete
+     *
+     * @param runs - how many times to run before returning
+     * @param counter - The counter provided from the caller to compare against
+     *
+     * @return The number of runs completed
+     */
+    @Asynchronous(runAt = @Schedule(cron = "*/3 * * * * *"))
+    public CompletableFuture<Integer> scheduledEvery3SecondsTakes5Seconds(final int runs, final AtomicInteger counter) {
+        int count = counter.incrementAndGet();
+
+        log.info("Executing scheduledEvery3SecondsTakes5Seconds method " + count + "/" + runs);
+        log.info("  Thread: " + Thread.currentThread().toString());
+
+        if (runs != count) {
+
+            try {
+                Thread.sleep(Duration.ofSeconds(5).toMillis());
+            } catch (InterruptedException e) {
+                throw new RuntimeException("Thread was interrupted while waiting", e);
+            }
+
+            return null; // Continue onto next scheduled execution
+        }
+
+        CompletableFuture<Integer> future = Asynchronous.Result.getFuture();
+        future.complete(count);
+
+        return future;
+    }
+
+    /**
+     * A scheduled async method that runs every 3 seconds
+     * Uses executor = "java:module/concurrent/ExecutorB" with max-async = 1
+     *
+     * @param runs - how many times to run before returning
+     * @param counter - The counter provided from the caller to compare against
+     *
+     * @return completed future of IntContext
+     */
+    @Asynchronous(executor = "java:module/concurrent/ExecutorB", runAt = @Schedule(cron = "*/3 * * * * *"))
+    public CompletableFuture<Integer> scheduledEvery3Seconds(final int runs, final AtomicInteger counter) {
+        int count = counter.incrementAndGet();
+
+        log.info("Executing scheduledEvery3Seconds method " + count + "/" + runs);
+        log.info("  Thread: " + Thread.currentThread().toString());
+
+        if (runs != count) {
+            return null; // Continue onto next scheduled execution
+        }
+
+        CompletableFuture<Integer> future = Asynchronous.Result.getFuture();
+        future.complete(IntContext.get());
+
+        return future;
+    }
+
+    /**
+     * A scheduled async method that runs every 3 seconds and every minute
+     * Uses executor = "java:app/concurrent/ExecutorA" with max-async = 1
+     *
+     * @param runs - how many times to run before returning
+     * @param counter - The counter provided from the caller to compare against
+     *
+     * @return completed future of StringContext
+     */
+    @Asynchronous(executor = "java:module/concurrent/ExecutorB", runAt = {
+            @Schedule(cron = "*/3 * * * * *"),
+            @Schedule(cron = "0 * * * * *")
+    })
+    public CompletableFuture<String> scheduledEvery3SecondsAnd1Minute(final int runs, final AtomicInteger counter) {
+        int count = counter.incrementAndGet();
+
+        log.info("Executing scheduledEvery3SecondsAnd1Minute method " + count + "/" + runs);
+        log.info("  Thread: " + Thread.currentThread().toString());
+
+        if (runs != count) {
+            return null; // Continue onto next scheduled execution
+        }
+
+        CompletableFuture<String> future = Asynchronous.Result.getFuture();
+        future.complete(StringContext.get());
+
+        return future;
+    }
+
+    /**
+     * A scheduled async method that should not run due to invalid configuration
+     */
+    @Asynchronous(executor = "java:app/concurrent/INVALID", runAt = @Schedule(cron = "*/3 * * * * *"))
+    public CompletableFuture<String> scheduledInvalidExecutor() {
+        throw new UnsupportedOperationException("Should not be able to execute with invalid executor");
+    }
 }

tck/src/main/java/ee/jakarta/tck/concurrent/spec/ManagedExecutorService/resourcedef/ManagedExecutorDefinitionFullTests.java

@@ -151,5 +151,46 @@ public void testManagedExecutorDefinitionAllAttributes() {
     public void testManagedExecutorDefinitionDefaults() {
         runTest(baseURL, testname);
     }
+
+    @Assertion(id = "GIT:439", strategy = "Ensure scheduled asynchronous methods are completed when future is completed.")
+    public void testScheduledAsynchCompletedFuture() {
+        runTest(baseURL, testname);
+    }
+
+    @Assertion(id = "GIT:439", strategy = "Ensure scheduled asynchronous methods are completed when a non-null result is returned.")
+    public void testScheduledAsynchCompletedResult() {
+        runTest(baseURL, testname);
+    }
+
+    @Assertion(id = "GIT:439", strategy = "Ensure scheduled asynchronous methods are completed when an exception is thrown.")
+    public void testScheduledAsynchCompletedExceptionally() {
+        runTest(baseURL, testname);
+    }
 
+    @Assertion(id = "GIT:439", strategy = "Ensure overlapping scheduled asynchronous methods are skipped.")
+    public void testScheduledAsynchOverlapSkipping() {
+        runTest(baseURL, testname);
+    }
+
+    @Assertion(id = "GIT:439", strategy = "Ensure scheduled asynchronous methods ignore the max-async configuration."
+            + " Ensure scheduled asynchronous methods honor cleared context configuration")
+    public void testScheduledAsynchIgnoresMaxAsync() {
+        runTest(baseURL, testname);
+    }
+
+    @Assertion(id = "GIT:439", strategy = "Ensure scheduled asynchronous methods choose closest execution time when multiple schedules are provided."
+            + " Ensure scheduled asynchronous methods honor propogated context configuration")
+    public void testScheduledAsynchWithMultipleSchedules() {
+        runTest(baseURL, testname);
+    }
+
+    @Assertion(id = "GIT:439", strategy = "Ensure scheduled asynchronous methods are not executed when an invalid JNDI name is provided.")
+    public void testScheduledAsynchWithInvalidJNDIName() {
+        runTest(baseURL, testname);
+    }
+
+    @Assertion(id = "GIT:439", strategy = "Ensure scheduled asynchronous methods with void return type stop execution via completable future or exception.")
+    public void testScheduledAsynchVoidReturn() {
+        runTest(baseURL, testname);
+    }
 }