Implement Concurrency 3.0

container/openejb-core/src/main/java/org/apache/openejb/cdi/CdiBeanInfo.java

@@ -28,6 +28,9 @@
 import org.apache.openejb.jee.JndiConsumer;
 import org.apache.openejb.jee.KeyedCollection;
 import org.apache.openejb.jee.LifecycleCallback;
+import org.apache.openejb.jee.ManagedExecutor;
+import org.apache.openejb.jee.ManagedScheduledExecutor;
+import org.apache.openejb.jee.ManagedThreadFactory;
 import org.apache.openejb.jee.MessageDestinationRef;
 import org.apache.openejb.jee.PersistenceContextRef;
 import org.apache.openejb.jee.PersistenceUnitRef;
@@ -67,6 +70,9 @@ public class CdiBeanInfo implements JndiConsumer {
     private ClassLoader classLoader;
     private List<Injection> injections;
     protected KeyedCollection<String, ContextService> contextService;
+    protected KeyedCollection<String, ManagedExecutor> managedExecutor;
+    protected KeyedCollection<String, ManagedScheduledExecutor> managedScheduledExecutor;
+    protected KeyedCollection<String, ManagedThreadFactory> managedThreadFactory;
 
     public String getBeanName() {
         return beanName;
@@ -323,11 +329,38 @@ public String getJndiConsumerName() {
     public Class<?> getBeanClass() {
         return this.beanClass;
     }
+
     @Override
     public Map<String, ContextService> getContextServiceMap() {
         if (contextService == null) {
             contextService = new KeyedCollection<String, ContextService>();
         }
         return this.contextService.toMap();
     }
+
+    @Override
+    public Map<String, ManagedExecutor> getManagedExecutorMap() {
+        if (managedExecutor == null) {
+            managedExecutor = new KeyedCollection<>();
+        }
+        return this.managedExecutor.toMap();
+    }
+
+    @Override
+    public Map<String, ManagedScheduledExecutor> getManagedScheduledExecutorMap() {
+        if (managedScheduledExecutor == null) {
+            managedScheduledExecutor = new KeyedCollection<>();
+        }
+
+        return this.managedScheduledExecutor.toMap();
+    }
+
+    @Override
+    public Map<String, ManagedThreadFactory> getManagedThreadFactoryMap() {
+        if (managedThreadFactory == null) {
+            managedThreadFactory = new KeyedCollection<>();
+        }
+
+        return this.managedThreadFactory.toMap();
+    }
 }

container/openejb-core/src/main/java/org/apache/openejb/cdi/CdiScanner.java

@@ -22,6 +22,7 @@
 import org.apache.openejb.assembler.classic.AppInfo;
 import org.apache.openejb.assembler.classic.BeansInfo;
 import org.apache.openejb.assembler.classic.EjbJarInfo;
+import org.apache.openejb.cdi.concurrency.AsynchronousInterceptor;
 import org.apache.openejb.cdi.transactional.MandatoryInterceptor;
 import org.apache.openejb.cdi.transactional.NeverInterceptor;
 import org.apache.openejb.cdi.transactional.NotSupportedInterceptor;
@@ -64,9 +65,13 @@ public class CdiScanner implements BdaScannerService {
     private static final Logger logger = Logger.getInstance(LogCategory.OPENEJB_CDI, OpenEJBLifecycle.class);
     public static final String OPENEJB_CDI_FILTER_CLASSLOADER = "openejb.cdi.filter.classloader";
 
-    private static final Class<?>[] TRANSACTIONAL_INTERCEPTORS = new Class<?>[]{
+    private static final Class<?>[] INTERNAL_INTERCEPTORS = new Class<?>[]{
+        // @Transactional
         MandatoryInterceptor.class, NeverInterceptor.class, NotSupportedInterceptor.class,
-        RequiredInterceptor.class, RequiredNewInterceptor.class, SupportsInterceptor.class
+        RequiredInterceptor.class, RequiredNewInterceptor.class, SupportsInterceptor.class,
+
+        // @Asynchronous
+        AsynchronousInterceptor.class
     };
 
     private final Set<Class<?>> startupClasses = new HashSet<>();
@@ -140,8 +145,8 @@ public void init(final Object object) {
 
             if (appInfo.webAppAlone || !ejbJar.webapp) {
                 // "manual" extension to avoid to add it through SPI mecanism
-                classes.addAll(asList(TRANSACTIONAL_INTERCEPTORS));
-                for (final Class<?> interceptor : TRANSACTIONAL_INTERCEPTORS) {
+                classes.addAll(asList(INTERNAL_INTERCEPTORS));
+                for (final Class<?> interceptor : INTERNAL_INTERCEPTORS) {
                     interceptorsManager.addEnabledInterceptorClass(interceptor);
                 }
             }

container/openejb-core/src/main/java/org/apache/openejb/cdi/ThreadSingletonServiceImpl.java

@@ -177,7 +177,7 @@ public void execute(final Runnable command) {
                                 executor = new ManagedExecutorServiceImpl(
                                         new ExecutorBuilder()
                                                 .size(3)
-                                                .threadFactory(new ManagedThreadFactoryImpl(appContext.getId() + "-cdi-fireasync-"))
+                                                .threadFactory(new ManagedThreadFactoryImpl(appContext.getId() + "-cdi-fireasync-", null, ContextServiceImplFactory.newPropagateEverythingContextService()))
                                                 .prefix("CDIAsyncPool")
                                                 .build(appContext.getOptions()), ContextServiceImplFactory.newPropagateEverythingContextService());
                                 delegate.compareAndSet(null, executor);

container/openejb-core/src/main/java/org/apache/openejb/cdi/concurrency/AsynchronousInterceptor.java

@@ -0,0 +1,115 @@
+/*
+ * 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.openejb.cdi.concurrency;
+
+import jakarta.annotation.Priority;
+import jakarta.enterprise.concurrent.Asynchronous;
+import jakarta.enterprise.concurrent.ManagedExecutorService;
+import jakarta.interceptor.AroundInvoke;
+import jakarta.interceptor.Interceptor;
+import jakarta.interceptor.InvocationContext;
+import org.apache.openejb.core.ivm.naming.NamingException;
+import org.apache.openejb.resource.thread.ManagedExecutorServiceImplFactory;
+
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.RejectedExecutionException;
+
+@Interceptor
+@Asynchronous
+@Priority(Interceptor.Priority.PLATFORM_BEFORE + 5)
+public class AsynchronousInterceptor {
+    public static final String MP_ASYNC_ANNOTATION_NAME = "org.eclipse.microprofile.faulttolerance.Asynchronous";
+
+    // ensure validation logic required by the spec only runs once per invoked Method
+    private final Map<Method, Exception> validationCache = new ConcurrentHashMap<>();
+
+    @AroundInvoke
+    public Object aroundInvoke(final InvocationContext ctx) throws Exception {
+        Exception exception = validationCache.computeIfAbsent(ctx.getMethod(), this::validate);
+        if (exception != null) {
+            throw exception;
+        }
+
+        Asynchronous asynchronous = ctx.getMethod().getAnnotation(Asynchronous.class);
+        ManagedExecutorService mes;
+        try {
+            mes = ManagedExecutorServiceImplFactory.lookup(asynchronous.executor());
+        } catch (NamingException | IllegalArgumentException e) {
+            throw new RejectedExecutionException("Cannot lookup ManagedExecutorService", e);
+        }
+
+        CompletableFuture<Object> future = mes.newIncompleteFuture();
+        mes.execute(() -> {
+            try {
+                Asynchronous.Result.setFuture(future);
+                CompletionStage<?> result = (CompletionStage<?>) ctx.proceed();
+                if (result == null || result == future) {
+                    future.complete(result);
+
+                    Asynchronous.Result.setFuture(null);
+                    return;
+                }
+
+                result.whenComplete((resultInternal, err) -> {
+                    if (resultInternal != null) {
+                        future.complete(resultInternal);
+                    } else if (err != null) {
+                        future.completeExceptionally(err);
+                    }
+
+                    Asynchronous.Result.setFuture(null);
+                });
+            } catch (Exception e) {
+                future.completeExceptionally(e);
+                Asynchronous.Result.setFuture(null);
+            }
+        });
+
+        return ctx.getMethod().getReturnType() == Void.TYPE ? null : future;
+    }
+
+    private Exception validate(final Method method) {
+        if (hasMpAsyncAnnotation(method.getAnnotations()) || hasMpAsyncAnnotation(method.getDeclaringClass().getAnnotations())) {
+            return new UnsupportedOperationException("Combining " + Asynchronous.class.getName()
+                    + " and " + MP_ASYNC_ANNOTATION_NAME + " on the same method/class is not supported");
+        }
+
+        Asynchronous asynchronous = method.getAnnotation(Asynchronous.class);
+        if (asynchronous == null) {
+            return new UnsupportedOperationException("Asynchronous annotation must be placed on a method");
+        }
+
+        Class<?> returnType = method.getReturnType();
+        if (returnType != Void.TYPE && returnType != CompletableFuture.class && returnType != CompletionStage.class) {
+            return new UnsupportedOperationException("Asynchronous annotation must be placed on a method that returns either void, CompletableFuture or CompletionStage");
+        }
+
+        return null;
+    }
+
+    private boolean hasMpAsyncAnnotation(Annotation[] declaredAnnotations) {
+        return Arrays.stream(declaredAnnotations)
+                .map(it -> it.annotationType().getName())
+                .anyMatch(it -> it.equals(MP_ASYNC_ANNOTATION_NAME));
+    }
+}

container/openejb-core/src/main/java/org/apache/openejb/config/AnnotationDeployer.java

@@ -18,6 +18,9 @@
 package org.apache.openejb.config;
 
 import jakarta.enterprise.concurrent.ContextServiceDefinition;
+import jakarta.enterprise.concurrent.ManagedExecutorDefinition;
+import jakarta.enterprise.concurrent.ManagedScheduledExecutorDefinition;
+import jakarta.enterprise.concurrent.ManagedThreadFactoryDefinition;
 import jakarta.interceptor.AroundConstruct;
 import org.apache.openejb.BeanContext;
 import org.apache.openejb.OpenEJBException;
@@ -78,6 +81,9 @@
 import org.apache.openejb.jee.Lifecycle;
 import org.apache.openejb.jee.LifecycleCallback;
 import org.apache.openejb.jee.Listener;
+import org.apache.openejb.jee.ManagedExecutor;
+import org.apache.openejb.jee.ManagedScheduledExecutor;
+import org.apache.openejb.jee.ManagedThreadFactory;
 import org.apache.openejb.jee.MessageAdapter;
 import org.apache.openejb.jee.MessageDrivenBean;
 import org.apache.openejb.jee.MessageListener;
@@ -4081,6 +4087,53 @@ public void buildAnnotatedRefs(final JndiConsumer consumer, final IAnnotationFin
                 buildContextServiceDefinition(consumer, definition);
             }
 
+            //
+            // @ManagedExecutorDefinition
+            //
+
+            for (final Annotated<Class<?>> annotated : annotationFinder.findMetaAnnotatedClasses(ManagedExecutorDefinition.List.class)) {
+                final ManagedExecutorDefinition.List defs = annotated.getAnnotation(ManagedExecutorDefinition.List.class);
+                for (final ManagedExecutorDefinition definition : defs.value()) {
+                    buildManagedExecutorDefinition(consumer, definition);
+                }
+            }
+
+            for (final Annotated<Class<?>> annotated : annotationFinder.findMetaAnnotatedClasses(ManagedExecutorDefinition.class)) {
+                final ManagedExecutorDefinition definition = annotated.getAnnotation(ManagedExecutorDefinition.class);
+                buildManagedExecutorDefinition(consumer, definition);
+            }
+
+            //
+            // @ManagedScheduledExecutorDefinition
+            //
+
+            for (final Annotated<Class<?>> annotated : annotationFinder.findMetaAnnotatedClasses(ManagedScheduledExecutorDefinition.List.class)) {
+                final ManagedScheduledExecutorDefinition.List defs = annotated.getAnnotation(ManagedScheduledExecutorDefinition.List.class);
+                for (final ManagedScheduledExecutorDefinition definition : defs.value()) {
+                    buildManagedScheduledExecutorDefinition(consumer, definition);
+                }
+            }
+
+            for (final Annotated<Class<?>> annotated : annotationFinder.findMetaAnnotatedClasses(ManagedScheduledExecutorDefinition.class)) {
+                final ManagedScheduledExecutorDefinition definition = annotated.getAnnotation(ManagedScheduledExecutorDefinition.class);
+                buildManagedScheduledExecutorDefinition(consumer, definition);
+            }
+
+            //
+            // @ManagedThreadFactoryDefinition
+            //
+            for (final Annotated<Class<?>> annotated : annotationFinder.findMetaAnnotatedClasses(ManagedThreadFactoryDefinition.List.class)) {
+                final ManagedThreadFactoryDefinition.List defs = annotated.getAnnotation(ManagedThreadFactoryDefinition.List.class);
+                for (final ManagedThreadFactoryDefinition definition : defs.value()) {
+                    buildManagedThreadFactoryDefinition(consumer, definition);
+                }
+            }
+
+            for (final Annotated<Class<?>> annotated : annotationFinder.findMetaAnnotatedClasses(ManagedThreadFactoryDefinition.class)) {
+                final ManagedThreadFactoryDefinition definition = annotated.getAnnotation(ManagedThreadFactoryDefinition.class);
+                buildManagedThreadFactoryDefinition(consumer, definition);
+            }
+
             //
             // @JMSConnectionFactoryDefinition
             //
@@ -4137,6 +4190,47 @@ private void buildContextServiceDefinition(final JndiConsumer consumer, final Co
             consumer.getContextServiceMap().put(definition.name(), contextService);
         }
 
+        private void buildManagedExecutorDefinition(final JndiConsumer consumer, final ManagedExecutorDefinition definition) {
+            ManagedExecutor existing = consumer.getManagedExecutorMap().get(definition.name());
+            final ManagedExecutor managedExecutor = (existing != null) ? existing : new ManagedExecutor();
+
+            managedExecutor.setName(new JndiName());
+            managedExecutor.getName().setvalue(definition.name());
+            managedExecutor.setContextService(new JndiName());
+            managedExecutor.getContextService().setvalue(definition.context());
+            managedExecutor.setHungTaskThreshold(definition.hungTaskThreshold());
+            managedExecutor.setMaxAsync(definition.maxAsync() == -1 ? null : definition.maxAsync());
+
+            consumer.getManagedExecutorMap().put(definition.name(), managedExecutor);
+        }
+
+        private void buildManagedScheduledExecutorDefinition(final JndiConsumer consumer, final ManagedScheduledExecutorDefinition definition) {
+            ManagedScheduledExecutor existing = consumer.getManagedScheduledExecutorMap().get(definition.name());
+            final ManagedScheduledExecutor managedScheduledExecutor = (existing != null) ? existing : new ManagedScheduledExecutor();
+
+            managedScheduledExecutor.setName(new JndiName());
+            managedScheduledExecutor.getName().setvalue(definition.name());
+            managedScheduledExecutor.setContextService(new JndiName());
+            managedScheduledExecutor.getContextService().setvalue(definition.context());
+            managedScheduledExecutor.setHungTaskThreshold(definition.hungTaskThreshold());
+            managedScheduledExecutor.setMaxAsync(definition.maxAsync() == -1 ? null : definition.maxAsync());
+
+            consumer.getManagedScheduledExecutorMap().put(definition.name(), managedScheduledExecutor);
+        }
+
+        private void buildManagedThreadFactoryDefinition(final JndiConsumer consumer, ManagedThreadFactoryDefinition definition) {
+            ManagedThreadFactory existing = consumer.getManagedThreadFactoryMap().get(definition.name());
+            final ManagedThreadFactory managedThreadFactory = (existing != null) ? existing : new ManagedThreadFactory();
+
+            managedThreadFactory.setName(new JndiName());
+            managedThreadFactory.getName().setvalue(definition.name());
+            managedThreadFactory.setContextService(new JndiName());
+            managedThreadFactory.getContextService().setvalue(definition.context());
+            managedThreadFactory.setPriority(definition.priority());
+
+            consumer.getManagedThreadFactoryMap().put(definition.name(), managedThreadFactory);
+        }
+
         private void buildContext(final JndiConsumer consumer, final Member member) {
             final ContextRef ref = new ContextRef();
             ref.setName(member.getDeclaringClass().getName() + "/" + member.getName());

container/openejb-core/src/main/java/org/apache/openejb/config/ConfigurationFactory.java

@@ -288,6 +288,9 @@ public ConfigurationFactory(final boolean offline, final DynamicDeployer preAuto
 
         chain.add(new ConvertDataSourceDefinitions());
         chain.add(new ConvertContextServiceDefinitions());
+        chain.add(new ConvertManagedExecutorServiceDefinitions());
+        chain.add(new ConvertManagedScheduledExecutorServiceDefinitions());
+        chain.add(new ConvertManagedThreadFactoryDefinitions());
         chain.add(new ConvertJMSConnectionFactoryDefinitions());
         chain.add(new ConvertJMSDestinationDefinitions());
         chain.add(new CleanEnvEntries());