Fixing swimlane dispatcher

eventuate-client-java-event-handling/build.gradle

@@ -8,6 +8,5 @@ dependencies {
 
     testCompile "junit:junit:4.11"
     testCompile 'org.mockito:mockito-core:1.9.5'
-
-
+    compile project(":eventuate-client-java-test-util")
 }

eventuate-client-java-event-handling/src/main/java/io/eventuate/javaclient/domain/SwimlaneDispatcher.java

@@ -28,6 +28,10 @@ public SwimlaneDispatcher(String subscriberId, Integer swimlane, Executor execut
     this.executor = executor;
   }
 
+  public boolean getRunning() {
+    return running.get();
+  }
+
   public CompletableFuture<?> dispatch(DispatchedEvent<Event> de, Function<DispatchedEvent<Event>, CompletableFuture<?>> target) {
     synchronized (queue) {
       QueuedEvent qe = new QueuedEvent(de, target);
@@ -59,26 +63,29 @@ public QueuedEvent(DispatchedEvent<Event> event, Function<DispatchedEvent<Event>
 
 
   public void processQueuedEvent() {
-    QueuedEvent qe = getNextEvent();
-    if (qe == null)
-      logger.trace("No queued event for {} {}", subscriberId, swimlane);
-    else {
-      logger.trace("Invoking handler for event for {} {} {}", subscriberId, swimlane, qe.event);
-      qe.target.apply(qe.event).handle((success, throwable) -> {
-        if (throwable == null) {
-          logger.debug("Handler succeeded for event for {} {} {}", subscriberId, swimlane, qe.event);
-          boolean x = qe.future.complete(success);
-          logger.trace("Completed future success {}", x);
-          logger.trace("Maybe processing next queued event {} {}", subscriberId, swimlane);
-          processNextQueuedEvent();
-        } else {
-          logger.error(String.format("handler for %s %s  %s failed: ", subscriberId, swimlane, qe.event), throwable);
-          boolean x = qe.future.completeExceptionally(throwable);
-          logger.trace("Completed future failed{}", x);
-          // TODO - what to do here???
-        }
-        return null;
-      });
+    while (true) {
+      QueuedEvent qe = getNextEvent();
+      if (qe == null) {
+        logger.trace("No queued event for {} {}", subscriberId, swimlane);
+        return;
+      }
+      else {
+        logger.trace("Invoking handler for event for {} {} {}", subscriberId, swimlane, qe.event);
+        qe.target.apply(qe.event).handle((success, throwable) -> {
+          if (throwable == null) {
+            logger.debug("Handler succeeded for event for {} {} {}", subscriberId, swimlane, qe.event);
+            boolean x = qe.future.complete(success);
+            logger.trace("Completed future success {}", x);
+            logger.trace("Maybe processing next queued event {} {}", subscriberId, swimlane);
+          } else {
+            logger.error(String.format("handler for %s %s  %s failed: ", subscriberId, swimlane, qe.event), throwable);
+            boolean x = qe.future.completeExceptionally(throwable);
+            logger.trace("Completed future failed{}", x);
+            // TODO - what to do here???
+          }
+          return null;
+        });
+      }
     }
   }
 

eventuate-client-java-event-handling/src/test/java/io/eventuate/javaclient/domain/SwimlaneDispatcherTest.java

@@ -0,0 +1,138 @@
+package io.eventuate.javaclient.domain;
+
+import io.eventuate.DispatchedEvent;
+import io.eventuate.Event;
+import io.eventuate.EventContext;
+import io.eventuate.Int128;
+import io.eventuate.testutil.Eventually;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.function.Function;
+
+public class SwimlaneDispatcherTest {
+  private SwimlaneDispatcher swimlaneDispatcher;
+  private Function<DispatchedEvent<Event>, CompletableFuture<?>> handler;
+  private Set<String> dataReturnedByHandlerFutures;
+  private Set<String> dataReceivedFromDispatcherFutures;
+  private Set<String> throwableDataReturnedByHandlerFutures;
+  private Set<String> throwableDataReceivedFromDispatcherFutures;
+
+  @Before
+  public void init() {
+    dataReturnedByHandlerFutures = new HashSet<>();
+    dataReceivedFromDispatcherFutures = new HashSet<>();
+    throwableDataReturnedByHandlerFutures = new HashSet<>();
+    throwableDataReceivedFromDispatcherFutures = new HashSet<>();
+
+    swimlaneDispatcher = new SwimlaneDispatcher("1", 1, Executors.newCachedThreadPool());
+  }
+
+  @Test
+  public void shouldDispatchManyEvents() {
+    int numberOfEventsToSend = 5;
+
+    createHandler(false);
+
+    sendEvents(numberOfEventsToSend);
+    assertEventsReceived(numberOfEventsToSend);
+  }
+
+  @Test
+  public void shouldHandleFailures() {
+    int numberOfEventsToSend = 5;
+
+    createHandler(true);
+
+    sendEvents(numberOfEventsToSend);
+    assertEventsFailed(numberOfEventsToSend);
+  }
+
+  @Test
+  public void testShouldRestart() {
+    int numberOfEventsToSend = 5;
+
+    createHandler(false);
+
+    sendEvents(numberOfEventsToSend);
+    assertDispatcherStopped();
+    sendEvents(numberOfEventsToSend);
+    assertEventsReceived(numberOfEventsToSend * 2);
+  }
+
+  private void createHandler(boolean shouldFail) {
+    handler = event -> {
+
+      try {
+        Thread.sleep(50);
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+
+      return CompletableFuture.supplyAsync(() -> {
+        String data = UUID.randomUUID().toString();
+
+        if (shouldFail) {
+          throwableDataReturnedByHandlerFutures.add(data);
+          throw new RuntimeException(data);
+        }
+
+        dataReturnedByHandlerFutures.add(data);
+        return data;
+      });
+    };
+  }
+
+  private void sendEvents(int numberOfEventsToSend) {
+    for (int i = 0; i < numberOfEventsToSend; i++) {
+      if (i > 0) {
+        Assert.assertTrue(swimlaneDispatcher.getRunning());
+      }
+      CompletableFuture<String> future = (CompletableFuture<String>)swimlaneDispatcher.dispatch(new DispatchedEvent<>("1",
+              new Int128(1, 1),
+              new Event() {},
+              1,
+              1L,
+              new EventContext(),
+              Optional.empty()), handler);
+
+      future.whenCompleteAsync((data, throwable) -> {
+        if (throwable != null) {
+          throwableDataReceivedFromDispatcherFutures.add(throwable.getCause().getMessage());
+        } else {
+          dataReceivedFromDispatcherFutures.add(data);
+        }
+      });
+    }
+  }
+
+  private void assertEventsReceived(int numberOfEventsToSend) {
+    Eventually.eventually(() -> {
+      Assert.assertTrue(throwableDataReturnedByHandlerFutures.isEmpty());
+      Assert.assertTrue(throwableDataReceivedFromDispatcherFutures.isEmpty());
+
+      Assert.assertEquals(numberOfEventsToSend, dataReturnedByHandlerFutures.size());
+      Assert.assertEquals(numberOfEventsToSend, dataReceivedFromDispatcherFutures.size());
+      Assert.assertEquals(dataReturnedByHandlerFutures, dataReceivedFromDispatcherFutures);
+    });
+  }
+
+  private void assertEventsFailed(int numberOfEventsToSend) {
+    Eventually.eventually(() -> {
+      Assert.assertTrue(dataReceivedFromDispatcherFutures.isEmpty());
+      Assert.assertTrue(dataReturnedByHandlerFutures.isEmpty());
+
+      Assert.assertEquals(numberOfEventsToSend, throwableDataReturnedByHandlerFutures.size());
+      Assert.assertEquals(numberOfEventsToSend, throwableDataReceivedFromDispatcherFutures.size());
+      Assert.assertEquals(throwableDataReturnedByHandlerFutures, throwableDataReceivedFromDispatcherFutures);
+    });
+  }
+
+  private void assertDispatcherStopped() {
+    Eventually.eventually(() -> Assert.assertFalse(swimlaneDispatcher.getRunning()));
+  }
+}