Fix for #14 - Using Spring AOP/@transactional on an event handler

eventuate-client-java-spring/src/main/java/io/eventuate/javaclient/spring/EventDispatcherInitializer.java

@@ -50,10 +50,10 @@ public EventDispatcherInitializer(EventHandlerProcessor[] processors, EventuateA
   }
 
 
-  public void registerEventHandler(Object eventHandlerBean, String beanName) {
+  public void registerEventHandler(Object eventHandlerBean, String beanName, Class<?> beanClass) {
 
-    List<AccessibleObject> fieldsAndMethods = Stream.<AccessibleObject>concat(Arrays.stream(ReflectionUtils.getUniqueDeclaredMethods(eventHandlerBean.getClass())),
-            Arrays.stream(eventHandlerBean.getClass().getDeclaredFields()))
+    List<AccessibleObject> fieldsAndMethods = Stream.<AccessibleObject>concat(Arrays.stream(ReflectionUtils.getUniqueDeclaredMethods(beanClass)),
+            Arrays.stream(beanClass.getDeclaredFields()))
             .collect(toList());
 
     List<AccessibleObject> annotatedCandidateEventHandlers = fieldsAndMethods.stream()
@@ -65,12 +65,15 @@ public void registerEventHandler(Object eventHandlerBean, String beanName) {
                     .process(eventHandlerBean, fieldOrMethod))
             .collect(toList());
 
+    if (handlers.isEmpty())
+      throw new RuntimeException("No handlers defined for this class" + beanClass);
+
     Map<String, Set<String>> aggregatesAndEvents = makeAggregatesAndEvents(handlers.stream()
             .filter(handler -> !handler.getEventType().equals(EndOfCurrentEventsReachedEvent.class)).collect(toList()));
 
     Map<Class<?>, EventHandler> eventTypesAndHandlers = makeEventTypesAndHandlers(handlers);
 
-    List<EventDeliveryExceptionHandler> exceptionHandlers = Arrays.stream(eventHandlerBean.getClass()
+    List<EventDeliveryExceptionHandler> exceptionHandlers = Arrays.stream(beanClass
             .getDeclaredFields())
             .filter(this::isExceptionHandlerField)
             .map(f -> {
@@ -83,7 +86,7 @@ public void registerEventHandler(Object eventHandlerBean, String beanName) {
             })
             .collect(toList());
 
-    EventSubscriber a = AnnotationUtils.findAnnotation(eventHandlerBean.getClass(), EventSubscriber.class);
+    EventSubscriber a = AnnotationUtils.findAnnotation(beanClass, EventSubscriber.class);
     if (a == null)
       throw new RuntimeException("Needs @EventSubscriber annotation: " + eventHandlerBean);
 
@@ -104,7 +107,7 @@ public void registerEventHandler(Object eventHandlerBean, String beanName) {
     try {
       aggregateStore.subscribe(subscriberId, aggregatesAndEvents,
               subscriberOptions, de -> swimlaneBasedDispatcher.dispatch(de, eventDispatcher::dispatch)).get(20, TimeUnit.SECONDS);
-      subscriptionsRegistry.add(new RegisteredSubscription(subscriberId, aggregatesAndEvents, eventHandlerBean.getClass()));
+      subscriptionsRegistry.add(new RegisteredSubscription(subscriberId, aggregatesAndEvents, beanClass));
     } catch (InterruptedException | TimeoutException | ExecutionException e) {
       throw new EventuateSubscriptionFailedException(subscriberId, e);
     }

eventuate-client-java-spring/src/main/java/io/eventuate/javaclient/spring/EventHandlerBeanPostProcessor.java

@@ -1,8 +1,10 @@
 package io.eventuate.javaclient.spring;
 
 import io.eventuate.EventSubscriber;
+import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.core.Ordered;
 import org.springframework.core.annotation.AnnotationUtils;
 
 /**
@@ -20,16 +22,15 @@ public EventHandlerBeanPostProcessor(EventDispatcherInitializer eventDispatcherI
 
   @Override
   public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
-    EventSubscriber a = AnnotationUtils.findAnnotation(bean.getClass(), EventSubscriber.class);
-    if (a != null)
-      eventDispatcherInitializer.registerEventHandler(bean, beanName);
     return bean;
   }
 
-
-
   @Override
   public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+    Class<?> actualClass = AopUtils.getTargetClass(bean);
+    EventSubscriber a = AnnotationUtils.findAnnotation(actualClass, EventSubscriber.class);
+    if (a != null)
+      eventDispatcherInitializer.registerEventHandler(bean, beanName, actualClass);
     return bean;
   }
 }

eventuate-client-java-tests-example/build.gradle

@@ -5,4 +5,7 @@ dependencies {
     compile project(":eventuate-client-java-spring")
     compile project(":eventuate-client-java-test-util")
     compile "io.reactivex:rxjava:$rxJavaVersion"
+    compile 'org.aspectj:aspectjrt:1.8.9'
+    compile 'org.aspectj:aspectjweaver:1.8.9'
+
 }

eventuate-client-java-tests-example/src/main/java/io/eventuate/example/banking/services/JavaIntegrationTestDomainConfiguration.java

@@ -4,10 +4,13 @@
 import io.eventuate.EventuateAggregateStore;
 import io.eventuate.example.banking.domain.Account;
 import io.eventuate.example.banking.domain.AccountCommand;
+import io.eventuate.example.banking.services.counting.InvocationCountingAspect;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
 
 @Configuration
+@EnableAspectJAutoProxy
 public class JavaIntegrationTestDomainConfiguration {
 
   @Bean
@@ -33,7 +36,12 @@ public AccountService accountService(AggregateRepository<Account, AccountCommand
 
   @Bean
   public AggregateRepository<Account, AccountCommand> accountRepository(EventuateAggregateStore aggregateStore) {
-    return new AggregateRepository<Account, AccountCommand>(Account.class, aggregateStore);
+    return new AggregateRepository<>(Account.class, aggregateStore);
+  }
+
+  @Bean
+  public InvocationCountingAspect loggingAspect() {
+    return new InvocationCountingAspect();
   }
 
 }

eventuate-client-java-tests-example/src/main/java/io/eventuate/example/banking/services/MoneyTransferCommandSideEventHandler.java

@@ -7,11 +7,13 @@
 import io.eventuate.EventSubscriber;
 import io.eventuate.example.banking.domain.AccountDebitedEvent;
 import io.eventuate.example.banking.domain.MoneyTransferCreatedEvent;
+import io.eventuate.example.banking.services.counting.Countable;
 import io.eventuate.testutil.AbstractTestEventHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 @EventSubscriber(id="javaIntegrationTestCommandSideMoneyTransferEventHandlers",progressNotifications = true)
+@Countable
 public class MoneyTransferCommandSideEventHandler extends AbstractTestEventHandler {
 
   private Logger logger = LoggerFactory.getLogger(getClass());

eventuate-client-java-tests-example/src/main/java/io/eventuate/example/banking/services/counting/Countable.java

@@ -0,0 +1,8 @@
+package io.eventuate.example.banking.services.counting;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Countable {
+}

eventuate-client-java-tests-example/src/main/java/io/eventuate/example/banking/services/counting/InvocationCountingAspect.java

@@ -0,0 +1,25 @@
+package io.eventuate.example.banking.services.counting;
+
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.annotation.Before;
+import org.aspectj.lang.annotation.Pointcut;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+@Aspect
+public class InvocationCountingAspect {
+
+  @Pointcut("@within(io.eventuate.example.banking.services.counting.Countable) && execution(public void *(..))")
+  public void invocation() {}
+
+  private AtomicLong counter = new AtomicLong(0);
+
+  @Before("invocation()")
+  public void countInvocation() {
+    counter.incrementAndGet();
+  }
+
+  public long getCounter() {
+    return counter.get();
+  }
+}

eventuate-client-java-tests-jdbc/src/test/java/io/eventuate/javaclient/spring/jdbc/JdbcAutoConfigurationIntegrationSyncTest.java

@@ -1,18 +1,28 @@
 package io.eventuate.javaclient.spring.jdbc;
 
-import io.eventuate.AggregateRepository;
+import io.eventuate.example.banking.services.counting.InvocationCountingAspect;
 import io.eventuate.javaclient.spring.tests.common.AbstractAccountIntegrationSyncTest;
-import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.IntegrationTest;
 import org.springframework.boot.test.SpringApplicationConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.Assert.assertTrue;
+
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringApplicationConfiguration(classes = JdbcAutoConfigurationIntegrationTestConfiguration.class)
 @IntegrationTest
 public class JdbcAutoConfigurationIntegrationSyncTest extends AbstractAccountIntegrationSyncTest {
 
+  @Autowired
+  private InvocationCountingAspect invocationCountingAspect;
 
+  @Override
+  public void shouldStartMoneyTransfer() throws ExecutionException, InterruptedException {
+    super.shouldStartMoneyTransfer();
+    assertTrue("Expected aspect to be called", invocationCountingAspect.getCounter() > 0);
+  }
 }

settings.gradle

@@ -18,7 +18,10 @@ include 'eventuate-client-java-embedded-test-aggregate-store-autoconfigure'
 include 'eventuate-client-java-tests-example-domain'
 include 'eventuate-client-java-tests-example'
 include 'eventuate-client-java-tests-common'
-include 'eventuate-client-java-tests'
+
+// Temporarily disable
+// include 'eventuate-client-java-tests'
+
 include 'eventuate-client-java-tests-jdbc'
 
 include 'eventuate-client-java-test-util'