Merge pull request #133 from dartartem/wip-db-id-gen

Extracted transaction manager from in-memory tests, added additional …

eventuate-tram-common-spring-in-memory/build.gradle

@@ -0,0 +1,6 @@
+apply plugin: PublicModulePlugin
+
+dependencies {
+    compile project(":eventuate-tram-in-memory")
+    compile "org.springframework.boot:spring-boot-starter-jdbc:$springBootCdcVersion"
+}

eventuate-tram-common-spring-in-memory/src/main/java/io/eventuate/tram/common/spring/inmemory/EventuateSpringTransactionSynchronizationManager.java

@@ -0,0 +1,24 @@
+package io.eventuate.tram.common.spring.inmemory;
+
+import io.eventuate.tram.inmemory.EventuateTransactionSynchronizationManager;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+public class EventuateSpringTransactionSynchronizationManager
+        implements EventuateTransactionSynchronizationManager {
+
+  @Override
+  public boolean isTransactionActive() {
+    return TransactionSynchronizationManager.isActualTransactionActive();
+  }
+
+  @Override
+  public void executeAfterTransaction(Runnable callback) {
+    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
+      @Override
+      public void afterCommit() {
+        callback.run();
+      }
+    });
+  }
+}

eventuate-tram-in-memory-test/build.gradle

@@ -0,0 +1,7 @@
+apply plugin: PublicModulePlugin
+
+dependencies {
+    compile project(":eventuate-tram-in-memory")
+
+    compile "junit:junit:4.12"
+}

eventuate-tram-in-memory-test/src/main/java/io/eventuate/tram/inmemory/test/AbstractInMemoryMessageProducerTest.java

@@ -0,0 +1,152 @@
+package io.eventuate.tram.inmemory.test;
+
+import io.eventuate.tram.messaging.common.Message;
+import io.eventuate.tram.messaging.consumer.MessageConsumer;
+import io.eventuate.tram.messaging.consumer.MessageHandler;
+import io.eventuate.tram.messaging.producer.MessageBuilder;
+import io.eventuate.tram.messaging.producer.MessageProducer;
+
+import java.util.Collections;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+public abstract class AbstractInMemoryMessageProducerTest {
+  protected String subscriberId;
+  protected String destination;
+  protected String payload;
+  protected MyMessageHandler mh;
+
+  protected abstract MessageProducer getMessageProducer();
+  protected abstract MessageConsumer getMessageConsumer();
+  protected abstract void executeInTransaction(Consumer<Runnable> callbackWithRollback);
+
+  public void setUp() {
+    subscriberId = "subscriberId-" + System.currentTimeMillis();
+    destination = "destination-" + System.currentTimeMillis();
+    payload = "payload-" + System.currentTimeMillis();
+    mh = new MyMessageHandler();
+  }
+
+  public void shouldDeliverToMatchingSubscribers() {
+    subscribe();
+    Message m = sendMessage();
+    assertNotNull(m.getId());
+    assertMessageReceived();
+  }
+
+  public void shouldSetIdWithinTransaction() {
+    Message m = makeMessage();
+    executeInTransaction(rollbackCallback -> {
+      getMessageProducer().send(destination, m);
+      assertNotNull(m.getId());
+    });
+  }
+
+  public void shouldDeliverToWildcardSubscribers() {
+    wildcardSubscribe();
+    sendMessage();
+    assertMessageReceived();
+  }
+
+  public void shouldReceiveMessageAfterTransaction() {
+    subscribe();
+    executeInTransaction(rollbackCallback -> sendMessage());
+    assertMessageReceived();
+  }
+
+  public void shouldNotReceiveMessageBeforeTransaction() {
+    subscribe();
+
+    executeInTransaction(rollbackCallback -> {
+      sendMessage();
+      assertMessageNotReceived();
+    });
+
+    assertMessageReceived();
+  }
+
+  public void shouldNotReceiveMessageAfterTransactionRollback() {
+    subscribe();
+
+    executeInTransaction(rollbackCallback -> {
+      sendMessage();
+      rollbackCallback.run();
+    });
+
+    assertMessageNotReceived();
+  }
+
+  protected void assertMessageReceived() {
+    mh.assertMessageReceived(payload);
+  }
+
+  protected void assertMessageNotReceived() {
+    mh.assertMessageNotReceived(payload);
+  }
+
+  protected Message sendMessage() {
+    Message m = makeMessage();
+    getMessageProducer().send(destination, m);
+    return m;
+  }
+
+  protected void wildcardSubscribe() {
+    subscribe("*");
+  }
+
+  protected void subscribe() {
+    subscribe(destination);
+  }
+
+  protected void subscribe(String destination) {
+    getMessageConsumer().subscribe(subscriberId, Collections.singleton(destination), mh);
+  }
+
+  protected Message makeMessage() {
+    return MessageBuilder.withPayload(payload).withHeader(Message.DESTINATION, destination).build();
+  }
+
+  protected static class MyMessageHandler implements MessageHandler {
+
+    protected BlockingQueue<String> queue = new LinkedBlockingDeque<>();
+
+    @Override
+    public void accept(Message message) {
+      try {
+        queue.put(message.getPayload());
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+    void assertMessageReceived(String payload) {
+      assertTrue(searchMessage(payload));
+    }
+
+    void assertMessageNotReceived(String payload) {
+      assertFalse(searchMessage(payload));
+    }
+
+    boolean searchMessage(String payload) {
+      String m;
+
+      try {
+        while ((m = queue.poll(3, TimeUnit.SECONDS)) != null) {
+          if (payload.equals(m)) {
+            return true;
+          }
+        }
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+
+      return false;
+    }
+  }
+}

eventuate-tram-in-memory/build.gradle

@@ -2,8 +2,8 @@ apply plugin: PublicModulePlugin
 
 dependencies {
     compile project(":eventuate-tram-messaging")
-    compile project(":eventuate-tram-spring-consumer-common")
-    compile project(":eventuate-tram-spring-messaging-producer-common")
+    compile project(":eventuate-tram-consumer-common")
+    compile project(":eventuate-tram-messaging-producer-common")
 
     compile project(":eventuate-tram-consumer-jdbc")
 

eventuate-tram-in-memory/src/main/java/io/eventuate/tram/inmemory/EventuateTransactionSynchronizationManager.java

@@ -0,0 +1,6 @@
+package io.eventuate.tram.inmemory;
+
+public interface EventuateTransactionSynchronizationManager {
+  boolean isTransactionActive();
+  void executeAfterTransaction(Runnable runnable);
+}

eventuate-tram-in-memory/src/main/java/io/eventuate/tram/inmemory/InMemoryMessageProducer.java

@@ -6,31 +6,27 @@
 import io.eventuate.tram.messaging.producer.common.MessageProducerImplementation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.transaction.support.TransactionSynchronizationAdapter;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 
 public class InMemoryMessageProducer implements MessageProducerImplementation {
 
   private Logger logger = LoggerFactory.getLogger(getClass());
 
   private final InMemoryMessageConsumer messageConsumer;
+  private EventuateTransactionSynchronizationManager eventuateTransactionSynchronizationManager;
 
   private ApplicationIdGenerator applicationIdGenerator = new ApplicationIdGenerator();
 
-  public InMemoryMessageProducer(InMemoryMessageConsumer messageConsumer) {
+  public InMemoryMessageProducer(InMemoryMessageConsumer messageConsumer,
+                                 EventuateTransactionSynchronizationManager eventuateTransactionSynchronizationManager) {
     this.messageConsumer = messageConsumer;
+    this.eventuateTransactionSynchronizationManager = eventuateTransactionSynchronizationManager;
   }
 
   @Override
   public void withContext(Runnable runnable) {
-    if (TransactionSynchronizationManager.isActualTransactionActive()) {
+    if (eventuateTransactionSynchronizationManager.isTransactionActive()) {
       logger.info("Transaction active");
-      TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
-        @Override
-        public void afterCommit() {
-          runnable.run();
-        }
-      });
+      eventuateTransactionSynchronizationManager.executeAfterTransaction(runnable);
     } else {
       logger.info("No transaction active");
       runnable.run();

eventuate-tram-micronaut-in-memory/build.gradle

@@ -14,6 +14,8 @@ apply plugin: PublicModulePlugin
 
 dependencies {
     compile project(":eventuate-tram-in-memory")
+    compile project(":eventuate-tram-common-spring-in-memory")
+    testCompile project(":eventuate-tram-in-memory-test")
 
     compile project(":eventuate-tram-micronaut-messaging")
     compile project(":eventuate-tram-micronaut-consumer-common")

eventuate-tram-micronaut-in-memory/src/main/java/io/eventuate/tram/micronaut/inmemory/TramInMemoryFactory.java

@@ -1,7 +1,7 @@
 package io.eventuate.tram.micronaut.inmemory;
 
-import io.eventuate.common.id.IdGenerator;
 import io.eventuate.common.inmemorydatabase.EventuateDatabaseScriptSupplier;
+import io.eventuate.tram.common.spring.inmemory.EventuateSpringTransactionSynchronizationManager;
 import io.eventuate.tram.consumer.common.MessageConsumerImplementation;
 import io.eventuate.tram.inmemory.InMemoryMessageConsumer;
 import io.eventuate.tram.inmemory.InMemoryMessageProducer;
@@ -33,7 +33,7 @@ public MessageConsumerImplementation messageConsumerImplementation(InMemoryMessa
 
   @Singleton
   public InMemoryMessageProducer inMemoryMessageProducer(InMemoryMessageConsumer messageConsumer) {
-    return new InMemoryMessageProducer(messageConsumer);
+    return new InMemoryMessageProducer(messageConsumer, new EventuateSpringTransactionSynchronizationManager());
   }
 
   @Singleton