pulled and merged

akka-spring/src/main/java/se/scalablesolutions/akka/spring/AkkaSpringInterceptor.java

@@ -0,0 +1,81 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.spring;
+
+import se.scalablesolutions.akka.actor.ActiveObjectAspect;
+import se.scalablesolutions.akka.actor.AspectInit;
+import se.scalablesolutions.akka.actor.AspectInitRegistry;
+import se.scalablesolutions.akka.actor.Dispatcher;
+
+import org.springframework.beans.factory.config.BeanPostProcessor;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+
+import java.lang.reflect.Method;
+
+public class AkkaSpringInterceptor extends ActiveObjectAspect implements MethodInterceptor, BeanPostProcessor {
+
+  static final String TRANSACTION_MANAGER_CLASS_NAME = "org.springframework.transaction.support.TransactionSynchronizationManager";
+  static final String IS_TRANSACTION_ALIVE_METHOD_NAME = "isActualTransactionActive";
+
+  static private Method IS_TRANSACTION_ALIVE_METHOD = null;
+
+  static {
+    try {
+      ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+      Class clazz = null;
+      if (contextClassLoader != null) {
+        clazz = contextClassLoader.loadClass(TRANSACTION_MANAGER_CLASS_NAME);
+      } else {
+        ClassLoader springClassLoader = AkkaSpringInterceptor.class.getClassLoader();
+        clazz = springClassLoader.loadClass(TRANSACTION_MANAGER_CLASS_NAME);
+      }
+      if (clazz != null) IS_TRANSACTION_ALIVE_METHOD = clazz.getDeclaredMethod(IS_TRANSACTION_ALIVE_METHOD_NAME, null);
+    } catch (Exception e) {
+    }
+  }
+
+  // FIXME make configurable
+  static final int TIME_OUT = 1000;
+
+  @Override
+  public Object invoke(MethodInvocation methodInvocation) throws Throwable {
+    Dispatcher dispatcher = new Dispatcher(isTransactional());
+    dispatcher.start();
+    try {
+      AspectInitRegistry.register(methodInvocation.getThis(), new AspectInit(
+          methodInvocation.getThis().getClass(),
+          dispatcher,
+          TIME_OUT));
+      return this.invoke(AkkaSpringJoinPointWrapper.createSpringAkkaAspectWerkzWrapper(methodInvocation));
+    } finally {
+      dispatcher.stop();
+    }
+  }
+
+  @Override
+  public Object postProcessAfterInitialization(Object bean, String arg) throws BeansException {
+    return bean;
+  }
+
+  @Override
+  public Object postProcessBeforeInitialization(Object bean, String arg) throws BeansException {
+    return bean;
+  }
+
+  /**
+   * Checks if intercepted Spring bean is in a transaction.
+   */
+  private boolean isTransactional() {
+    try {
+      return (Boolean) IS_TRANSACTION_ALIVE_METHOD.invoke(null, null);
+    } catch (Exception e) {
+      throw new RuntimeException("Could not check if the Spring bean is executing within a transaction", e);
+    }
+  }
+}

akka-spring/src/main/java/se/scalablesolutions/akka/spring/AkkaSpringJoinPointWrapper.java

@@ -0,0 +1,89 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.spring;
+
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.codehaus.aspectwerkz.joinpoint.*;
+import org.codehaus.aspectwerkz.joinpoint.management.JoinPointType;
+
+public class AkkaSpringJoinPointWrapper implements JoinPoint {
+
+  private MethodInvocation methodInvocation = null;
+
+  public static AkkaSpringJoinPointWrapper createSpringAkkaAspectWerkzWrapper(MethodInvocation methodInvocation) {
+    AkkaSpringJoinPointWrapper joinPointWrapper = new AkkaSpringJoinPointWrapper();
+    joinPointWrapper.setMethodInvocation(methodInvocation);
+    return joinPointWrapper;
+  }
+
+  public MethodInvocation getMethodInvocation() {
+    return methodInvocation;
+  }
+
+  public void setMethodInvocation(MethodInvocation methodInvocation) {
+    this.methodInvocation = methodInvocation;
+  }
+
+  public Object proceed() throws Throwable {
+    return methodInvocation.proceed();
+  }
+
+  public Rtti getRtti() {
+    return new AkkaSpringRttiWrapper(methodInvocation);
+  }
+
+  public Object getTarget() {
+    return methodInvocation.getThis();
+  }
+
+  public Object getThis() {
+    return methodInvocation.getThis();
+  }
+
+  public Object getCallee() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Object getCaller() {
+    throw new UnsupportedOperationException();
+  }
+
+  public void addMetaData(Object arg0, Object arg1) {
+    throw new UnsupportedOperationException();
+  }
+
+  public Class getCalleeClass() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Class getCallerClass() {
+    throw new UnsupportedOperationException();
+  }
+
+  public EnclosingStaticJoinPoint getEnclosingStaticJoinPoint() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Object getMetaData(Object arg0) {
+    throw new UnsupportedOperationException();
+  }
+
+  public Signature getSignature() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Class getTargetClass() {
+    throw new UnsupportedOperationException();
+  }
+
+  public JoinPointType getType() {
+    throw new UnsupportedOperationException();
+  }
+
+  public StaticJoinPoint copy() {
+    throw new UnsupportedOperationException();
+  }
+}

akka-spring/src/main/java/se/scalablesolutions/akka/spring/AkkaSpringRttiWrapper.java

@@ -0,0 +1,73 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.spring;
+
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.codehaus.aspectwerkz.joinpoint.MethodRtti;
+import org.codehaus.aspectwerkz.joinpoint.Rtti;
+
+import java.lang.reflect.Method;
+
+public class AkkaSpringRttiWrapper implements MethodRtti {
+
+  private MethodInvocation methodInvocation = null;
+
+  public AkkaSpringRttiWrapper(MethodInvocation methodInvocation) {
+    this.methodInvocation = methodInvocation;
+  }
+
+  public Method getMethod() {
+    return methodInvocation.getMethod();
+  }
+
+  public Class getReturnType() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Object getReturnValue() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Class[] getExceptionTypes() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Class[] getParameterTypes() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Object[] getParameterValues() {
+    throw new UnsupportedOperationException();
+  }
+
+  public void setParameterValues(Object[] arg0) {
+    throw new UnsupportedOperationException();
+  }
+
+  public Rtti cloneFor(Object arg0, Object arg1) {
+    throw new UnsupportedOperationException();
+  }
+
+  public Class getDeclaringType() {
+    throw new UnsupportedOperationException();
+  }
+
+  public int getModifiers() {
+    throw new UnsupportedOperationException();
+  }
+
+  public String getName() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Object getTarget() {
+    throw new UnsupportedOperationException();
+  }
+
+  public Object getThis() {
+    throw new UnsupportedOperationException();
+  }
+}

akka-spring/src/test/java/se/scalablesolutions/akka/spring/AkkaSpringInterceptorTest.java

@@ -0,0 +1,44 @@
+package se.scalablesolutions.akka.spring;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+import junit.framework.TestSuite;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+public class AkkaSpringInterceptorTest extends TestCase {
+
+  public AkkaSpringInterceptorTest(String testName) {
+    super(testName);
+  }
+
+  public static Test suite() {
+    return new TestSuite(AkkaSpringInterceptorTest.class);
+  }
+
+  public void testInvokingAkkaEnabledSpringBeanMethodWithReturnValue() {
+    ApplicationContext context = new ClassPathXmlApplicationContext("spring-test-config.xml");
+    MyService myService = (MyService) context.getBean("actorBeanService");
+    Object obj = myService.getNumbers(12);
+    assertEquals(new Integer(12), obj);
+  }
+
+  public void testThatAkkaEnabledSpringBeanMethodCallIsInvokedInADifferentThreadThanTheTest() {
+    ApplicationContext context = new ClassPathXmlApplicationContext("spring-test-config.xml");
+    MyService myService = (MyService) context.getBean("actorBeanService");
+    assertNotSame(Thread.currentThread().getName(), myService.getThreadName());
+  }
+
+  public void testInvokingAkkaEnabledSpringBeanMethodWithTransactionalMethod() {
+    ApplicationContext context = new ClassPathXmlApplicationContext("spring-test-config.xml");
+    MyService myService = (MyService) context.getBean("actorBeanService");
+    myService.init();
+    myService.setMapState("key", "value");
+    myService.setRefState("value");
+    myService.setVectorState("value");
+    assertEquals("value", myService.getMapState("key"));
+    assertEquals("value", myService.getRefState());
+    assertEquals("value", myService.getVectorState());
+  }
+}

akka-spring/src/test/java/se/scalablesolutions/akka/spring/MyService.java

@@ -0,0 +1,79 @@
+package se.scalablesolutions.akka.spring;
+
+import org.springframework.transaction.annotation.Transactional;
+
+import se.scalablesolutions.akka.state.TransactionalState;
+import se.scalablesolutions.akka.state.TransactionalRef;
+import se.scalablesolutions.akka.state.TransactionalVector;
+import se.scalablesolutions.akka.state.TransactionalMap;
+
+import se.scalablesolutions.akka.annotation.oneway;
+import se.scalablesolutions.akka.annotation.prerestart;
+import se.scalablesolutions.akka.annotation.postrestart;
+
+public class MyService {
+
+  private TransactionalMap<String, String> mapState;
+  private TransactionalVector<String> vectorState;
+  private TransactionalRef<String> refState;
+  private boolean isInitialized = false;
+
+  @Transactional
+  public void init() {
+    if (!isInitialized) {
+      mapState = TransactionalState.newMap();
+      vectorState = TransactionalState.newVector();
+      refState = TransactionalState.newRef();
+      isInitialized = true;
+    }
+  }
+
+  public String getThreadName() {
+    return Thread.currentThread().getName();
+  }
+
+  public Integer getNumbers(int aTestNumber) {
+    return new Integer(aTestNumber);
+  }
+  
+  @Transactional
+  public String getMapState(String key) {
+    return (String)mapState.get(key).get();
+  }
+
+  @Transactional
+  public String getVectorState() {
+    return (String)vectorState.last();
+  }
+
+  @Transactional
+  public String getRefState() {
+    return (String)refState.get().get();
+  }
+
+  @Transactional
+  public void setMapState(String key, String msg) {
+    mapState.put(key, msg);
+  }
+
+  @Transactional
+  public void setVectorState(String msg) {
+    vectorState.add(msg);
+  }
+
+  @Transactional
+  public void setRefState(String msg) {
+    refState.swap(msg);
+  }
+
+  @prerestart
+  public void preRestart() {
+    System.out.println("################ PRE RESTART");
+  }
+
+  @postrestart
+  public void postRestart() {
+    System.out.println("################ POST RESTART");
+  }
+}
+

akka-spring/src/test/resources/spring-test-config.xml

@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
+
+  <bean id="akkaInterceptor" class="se.scalablesolutions.akka.spring.AkkaSpringInterceptor">
+  </bean>
+
+  <bean id="service" class="se.scalablesolutions.akka.spring.MyService">
+  </bean>
+
+  <bean id="actorBeanService" class="org.springframework.aop.framework.ProxyFactoryBean">
+    <property name="target">
+      <ref bean="service"/>
+    </property>
+    <property name="interceptorNames">
+      <list>
+        <value>akkaInterceptor</value>
+      </list>
+    </property>
+  </bean>
+
+</beans>

pom.xml

@@ -59,7 +59,6 @@
     <module>akka-comet</module>
     <module>akka-amqp</module>
     <module>akka-security</module>
-    <module>akka-patterns</module>
     <module>akka-kernel</module>
     <module>akka-fun-test-java</module>
     <module>akka-samples</module>