ISPN-1636 Have remote tasks inject services via CDI before execution

cdi/extension/src/main/java/org/infinispan/cdi/BeanManagerController.java

@@ -0,0 +1,95 @@
+/* 
+ * JBoss, Home of Professional Open Source 
+ * Copyright 2012 Red Hat Inc. and/or its affiliates and other contributors
+ * as indicated by the @author tag. All rights reserved. 
+ * See the copyright.txt in the distribution for a 
+ * full listing of individual contributors.
+ *
+ * This copyrighted material is made available to anyone wishing to use, 
+ * modify, copy, or redistribute it subject to the terms and conditions 
+ * of the GNU Lesser General Public License, v. 2.1. 
+ * This program is distributed in the hope that it will be useful, but WITHOUT A 
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A 
+ * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details. 
+ * You should have received a copy of the GNU Lesser General Public License, 
+ * v.2.1 along with this distribution; if not, write to the Free Software 
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, 
+ * MA  02110-1301, USA.
+ */
+package org.infinispan.cdi;
+
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import javax.enterprise.inject.spi.BeanManager;
+
+import org.jboss.solder.beanManager.BeanManagerAware;
+
+public class BeanManagerController extends BeanManagerAware {
+
+   private Map<ClassLoader, BeanManager> beanManagers = new ConcurrentHashMap<ClassLoader, BeanManager>();
+
+   public BeanManager getRegisteredBeanManager() {
+      ClassLoader classLoader = getClassLoader(null);
+
+      BeanManager bm = resolveBeanManager(classLoader);
+
+      if (bm == null) {
+         bm = getBeanManager();
+
+         if (bm == null) {
+            throw new IllegalStateException("Can not find BeanManager. Check CDI setup in your execution environment!");
+         }
+
+         registerBeanManager(classLoader, bm);
+
+      }
+      return bm;
+   }
+
+   public void registerBeanManager(ClassLoader cl, BeanManager bm){
+      beanManagers.put(cl, bm);
+   }
+
+   public void registerBeanManager(BeanManager bm){
+      beanManagers.put(getClassLoader(null), bm);
+   }
+
+   public void deregisterBeanManager(ClassLoader cl){
+      beanManagers.remove(cl);
+   }
+
+   public void deregisterBeanManager(){
+      beanManagers.remove(getClassLoader(null));
+   }
+
+   public BeanManager resolveBeanManager(ClassLoader cl){
+      return beanManagers.get(cl);
+   }
+
+   public ClassLoader getClassLoader(Object o) {
+      ClassLoader loader = AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() {
+
+         @Override
+         public ClassLoader run() {
+            try {
+               return Thread.currentThread().getContextClassLoader();
+            } catch (Exception e) {
+               return null;
+            }
+         }
+      });
+
+      if (loader == null && o != null) {
+         loader = o.getClass().getClassLoader();
+      }
+
+      if (loader == null) {
+         loader = BeanManagerController.class.getClassLoader();
+      }
+
+      return loader;
+   }
+}

cdi/extension/src/main/java/org/infinispan/cdi/CDIDistributedTaskLifecycle.java

@@ -0,0 +1,37 @@
+package org.infinispan.cdi;
+
+import java.util.concurrent.Callable;
+
+import javax.enterprise.context.spi.CreationalContext;
+import javax.enterprise.inject.spi.AnnotatedType;
+import javax.enterprise.inject.spi.BeanManager;
+import javax.enterprise.inject.spi.InjectionTarget;
+
+import org.infinispan.distexec.spi.DistributedTaskLifecycle;
+import org.jboss.solder.beanManager.BeanManagerAware;
+
+public class CDIDistributedTaskLifecycle extends BeanManagerAware implements DistributedTaskLifecycle {
+
+   @Override
+   @SuppressWarnings({ "unchecked" })
+   public <T> void onPreExecute(Callable<T> task) {
+      BeanManager bm = InfinispanExtension.getBeanManagerController().getRegisteredBeanManager();
+      Class<Callable<T>> clazz = (Class<Callable<T>>) task.getClass();
+      AnnotatedType<Callable<T>> type = bm.createAnnotatedType(clazz);
+      InjectionTarget<Callable<T>> it = bm.createInjectionTarget(type);
+      CreationalContext<Callable<T>> ctx = bm.createCreationalContext(null);
+      it.inject(task, ctx);
+      it.postConstruct(task);
+   }
+
+   @Override
+   @SuppressWarnings({ "unchecked" })
+   public <T> void onPostExecute(Callable<T> task) {            
+      BeanManager bm = InfinispanExtension.getBeanManagerController().getRegisteredBeanManager();
+      Class<Callable<T>> clazz = (Class<Callable<T>>) task.getClass();
+      AnnotatedType<Callable<T>> type = bm.createAnnotatedType(clazz);
+      InjectionTarget<Callable<T>> it = bm.createInjectionTarget(type);
+      it.preDestroy(task);
+      it.dispose(task);
+   }
+}

cdi/extension/src/main/java/org/infinispan/cdi/CDIMapReduceTaskLifecycle.java

@@ -0,0 +1,60 @@
+package org.infinispan.cdi;
+
+import javax.enterprise.context.spi.CreationalContext;
+import javax.enterprise.inject.spi.AnnotatedType;
+import javax.enterprise.inject.spi.BeanManager;
+import javax.enterprise.inject.spi.InjectionTarget;
+
+import org.infinispan.distexec.mapreduce.Mapper;
+import org.infinispan.distexec.mapreduce.Reducer;
+import org.infinispan.distexec.mapreduce.spi.MapReduceTaskLifecycle;
+import org.jboss.solder.beanManager.BeanManagerAware;
+
+public class CDIMapReduceTaskLifecycle extends BeanManagerAware implements MapReduceTaskLifecycle {
+
+   @Override
+   @SuppressWarnings({ "unchecked" })
+   public <KIn, VIn, KOut, VOut> void onPreExecute(Mapper<KIn, VIn, KOut, VOut> mapper) {
+      BeanManager bm = InfinispanExtension.getBeanManagerController().getRegisteredBeanManager();
+      Class<Mapper<KIn, VIn, KOut, VOut>> clazz = (Class<Mapper<KIn, VIn, KOut, VOut>>) mapper.getClass();
+      AnnotatedType<Mapper<KIn, VIn, KOut, VOut>> type = bm.createAnnotatedType(clazz);
+      InjectionTarget<Mapper<KIn, VIn, KOut, VOut>> it = bm.createInjectionTarget(type);
+      CreationalContext<Mapper<KIn, VIn, KOut, VOut>> ctx = bm.createCreationalContext(null);
+      it.inject(mapper, ctx);
+      it.postConstruct(mapper);
+   }
+
+   @Override
+   @SuppressWarnings({ "unchecked" })
+   public <KIn, VIn, KOut, VOut> void onPostExecute(Mapper<KIn, VIn, KOut, VOut> mapper) {
+      BeanManager bm = InfinispanExtension.getBeanManagerController().getRegisteredBeanManager();
+      Class<Mapper<KIn, VIn, KOut, VOut>> clazz = (Class<Mapper<KIn, VIn, KOut, VOut>>) mapper.getClass();
+      AnnotatedType<Mapper<KIn, VIn, KOut, VOut>> type = bm.createAnnotatedType(clazz);
+      InjectionTarget<Mapper<KIn, VIn, KOut, VOut>> it = bm.createInjectionTarget(type);
+      it.preDestroy(mapper);
+      it.dispose(mapper);
+   }
+
+   @Override
+   @SuppressWarnings({ "unchecked" })
+   public <KOut, VOut> void onPreExecute(Reducer<KOut, VOut> reducer) {
+      BeanManager bm = InfinispanExtension.getBeanManagerController().getRegisteredBeanManager();
+      Class<Reducer<KOut, VOut>> clazz = (Class<Reducer<KOut, VOut>>) reducer.getClass();
+      AnnotatedType<Reducer<KOut, VOut>> type = bm.createAnnotatedType(clazz);
+      InjectionTarget<Reducer<KOut, VOut>> it = bm.createInjectionTarget(type);
+      CreationalContext<Reducer<KOut, VOut>> ctx = bm.createCreationalContext(null);
+      it.inject(reducer, ctx);
+      it.postConstruct(reducer);
+   }
+
+   @Override
+   @SuppressWarnings({ "unchecked" })
+   public <KOut, VOut> void onPostExecute(Reducer<KOut, VOut> reducer) {
+      BeanManager bm = InfinispanExtension.getBeanManagerController().getRegisteredBeanManager();
+      Class<Reducer<KOut, VOut>> clazz = (Class<Reducer<KOut, VOut>>) reducer.getClass();
+      AnnotatedType<Reducer<KOut, VOut>> type = bm.createAnnotatedType(clazz);
+      InjectionTarget<Reducer<KOut, VOut>> it = bm.createInjectionTarget(type);
+      it.preDestroy(reducer);
+      it.dispose(reducer);
+   }
+}

cdi/extension/src/main/java/org/infinispan/cdi/InfinispanExtension.java

@@ -39,6 +39,7 @@
 import org.infinispan.util.logging.LogFactory;
 import org.jboss.solder.bean.BeanBuilder;
 import org.jboss.solder.bean.ContextualLifecycle;
+import org.jboss.solder.beanManager.BeanManagerAware;
 import org.jboss.solder.reflection.annotated.AnnotatedTypeBuilder;
 
 import javax.cache.annotation.CachePut;
@@ -56,6 +57,7 @@
 import javax.enterprise.inject.spi.Bean;
 import javax.enterprise.inject.spi.BeanManager;
 import javax.enterprise.inject.spi.BeforeBeanDiscovery;
+import javax.enterprise.inject.spi.BeforeShutdown;
 import javax.enterprise.inject.spi.Extension;
 import javax.enterprise.inject.spi.InjectionPoint;
 import javax.enterprise.inject.spi.InjectionTarget;
@@ -70,7 +72,6 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
-
 import static org.jboss.solder.bean.Beans.getQualifiers;
 import static org.jboss.solder.reflection.AnnotationInspector.getMetaAnnotation;
 import static org.jboss.solder.reflection.Reflections.getRawType;
@@ -81,14 +82,14 @@
  * @author Pete Muir
  * @author Kevin Pollet <kevin.pollet@serli.com> (C) 2011 SERLI
  */
-public class InfinispanExtension implements Extension {
-
-   private static final Log log = LogFactory.getLog(InfinispanExtension.class, Log.class);
+public class InfinispanExtension extends BeanManagerAware implements Extension {
 
    private Producer<RemoteCache<?, ?>> remoteCacheProducer;
+   private static final Log log = LogFactory.getLog(InfinispanExtension.class, Log.class);  
+   private static BeanManagerController bmc;
+
    private final Set<ConfigurationHolder> configurations;
-   private final Map<Type, Set<Annotation>> remoteCacheInjectionPoints;
-
+   private final Map<Type, Set<Annotation>> remoteCacheInjectionPoints;   
    InfinispanExtension() {
       this.configurations = new HashSet<InfinispanExtension.ConfigurationHolder>();
       this.remoteCacheInjectionPoints = new HashMap<Type, Set<Annotation>>();
@@ -229,6 +230,24 @@ void registerCacheConfigurations(@Observes AfterDeploymentValidation event, Cach
          eventBridge.registerObservers(cacheQualifiers, cacheName, cacheManager);
       }
    }
+
+   public static BeanManagerController getBeanManagerController() {
+      if (bmc == null) {
+         throw new IllegalStateException("CDI not properly set up in your execution environment!");
+      }
+      return bmc;
+   }
+
+   protected void setBeanManager(@Observes AfterBeanDiscovery afterBeanDiscovery, BeanManager beanManager) {      
+      if (bmc == null) {        
+         bmc = new BeanManagerController();
+      }
+      bmc.registerBeanManager(beanManager);
+   }
+
+   protected void cleanupBeanManager(@Observes BeforeShutdown beforeShutdown) {      
+      bmc.deregisterBeanManager();
+   }
 
    static class ConfigurationHolder {
       private final Producer<Configuration> producer;

cdi/extension/src/main/resources/META-INF/services/org.infinispan.distexec.mapreduce.spi.MapReduceTaskLifecycle

@@ -0,0 +1 @@
+org.infinispan.cdi.CDIMapReduceTaskLifecycle

cdi/extension/src/main/resources/META-INF/services/org.infinispan.distexec.spi.DistributedTaskLifecycle

@@ -0,0 +1 @@
+org.infinispan.cdi.CDIDistributedTaskLifecycle

cdi/extension/src/test/java/org/infinispan/cdi/test/distexec/DistributedExecutorCDITest.java

@@ -0,0 +1,83 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2011 Red Hat Inc. and/or its affiliates and other
+ * contributors as indicated by the @author tags. All rights reserved.
+ * See the copyright.txt in the distribution for a full listing of
+ * individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.infinispan.cdi.test.distexec;
+
+import static org.infinispan.cdi.test.testutil.Deployments.baseDeployment;
+
+import java.io.Serializable;
+import java.util.concurrent.Callable;
+
+import javax.inject.Inject;
+
+import org.infinispan.Cache;
+import org.infinispan.distexec.DistributedExecutorTest;
+import org.infinispan.test.MultipleCacheManagersTest;
+import org.jboss.arquillian.container.test.api.Deployment;
+import org.jboss.shrinkwrap.api.Archive;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+/**
+ * Tests CDI integration with org.infinispan.distexec.DistributedExecutorService
+ * 
+ * @author Vladimir Blagojevic 
+ */
+@Test(enabled = true, groups = "functional", testName = "distexec.DistributedExecutorTest")
+public class DistributedExecutorCDITest extends MultipleCacheManagersArquillianTest {
+
+   DistributedExecutorTest delegate;
+
+   public DistributedExecutorCDITest() {
+      delegate = new DistributedExecutorTest();
+   }
+
+   @Override
+   MultipleCacheManagersTest getDelegate() {
+      return delegate;
+   }
+
+   @Deployment
+   public static Archive<?> deployment() {
+      return baseDeployment().addClass(DistributedExecutorCDITest.class);
+   }
+
+   public void testBasicInvocation() throws Exception {
+      delegate.basicInvocation(new SimpleCallable());
+   } 
+
+
+   static class SimpleCallable implements Callable<Integer>, Serializable {
+
+      /** The serialVersionUID */
+      private static final long serialVersionUID = -8589149500259272402L;
+
+      @Inject
+      private Cache<String, String> cache;
+
+      @Override
+      public Integer call() throws Exception {
+         Assert.assertNotNull(cache, "Cache not injected into " + this);
+         return 1;
+      }
+   }
+}