Add more control over BatchEventProcessor. (#319)

* Add flush method to BatchEventProcessor.
* Allow BatchEventProcessor to be restarted.
* Add option to build BatchEventProcessor without starting the processing thread.

Author: Mike Davis

core-api/src/main/java/com/optimizely/ab/event/BatchEventProcessor.java

@@ -50,6 +50,7 @@ public class BatchEventProcessor implements EventProcessor, AutoCloseable {
     public static final long DEFAULT_TIMEOUT_INTERVAL = TimeUnit.SECONDS.toMillis(5);
 
     private static final Object SHUTDOWN_SIGNAL = new Object();
+    private static final Object FLUSH_SIGNAL    = new Object();
 
     private final BlockingQueue<Object> eventQueue;
     private final EventHandler eventHandler;
@@ -81,8 +82,6 @@ private BatchEventProcessor(BlockingQueue<Object> eventQueue, EventHandler event
         } else {
             this.executor = executor;
         }
-
-        start();
     }
 
     public synchronized void start() {
@@ -107,6 +106,8 @@ public void close() throws Exception {
             Thread.currentThread().interrupt();
         } catch (TimeoutException e) {
             logger.error("Timeout exceeded attempting to close for {} ms", timeoutMillis);
+        } finally {
+            isStarted = false;
         }
     }
 
@@ -123,6 +124,10 @@ public void process(UserEvent userEvent) {
         }
     }
 
+    public void flush() throws InterruptedException {
+        eventQueue.put(FLUSH_SIGNAL);
+    }
+
     public class EventConsumer implements Runnable {
         private LinkedList<UserEvent> currentBatch = new LinkedList<>();
         private long deadline = System.currentTimeMillis() + flushInterval;
@@ -148,6 +153,12 @@ public void run() {
                         break;
                     }
 
+                    if (item == FLUSH_SIGNAL) {
+                        logger.debug("Received flush signal.");
+                        flush();
+                        continue;
+                    }
+
                     addToBatch((UserEvent) item);
                 }
             } catch (InterruptedException e) {
@@ -267,8 +278,17 @@ public Builder withNotificationCenter(NotificationCenter notificationCenter) {
         }
 
         public BatchEventProcessor build() {
-            return new BatchEventProcessor(eventQueue, eventHandler, batchSize, flushInterval, timeoutMillis, executor, notificationCenter);
+            return build(true);
         }
-    }
 
+        public BatchEventProcessor build(boolean shouldStart) {
+            BatchEventProcessor batchEventProcessor = new BatchEventProcessor(eventQueue, eventHandler, batchSize, flushInterval, timeoutMillis, executor, notificationCenter);
+
+            if (shouldStart) {
+                batchEventProcessor.start();
+            }
+
+            return batchEventProcessor;
+        }
+    }
 }

core-api/src/test/java/com/optimizely/ab/event/BatchEventProcessorTest.java

@@ -121,6 +121,28 @@ public void testFlushMaxBatchSize() throws Exception {
         assertEquals(0, eventQueue.size());
     }
 
+    @Test
+    public void testFlush() throws Exception {
+        CountDownLatch countDownLatch = new CountDownLatch(2);
+        setEventProcessor(logEvent -> {
+            eventHandlerRule.dispatchEvent(logEvent);
+            countDownLatch.countDown();
+        });
+
+        UserEvent userEvent = buildConversionEvent(EVENT_NAME);
+        eventProcessor.process(userEvent);
+        eventProcessor.flush();
+        eventHandlerRule.expectConversion(EVENT_NAME, USER_ID);
+
+        eventProcessor.process(userEvent);
+        eventProcessor.flush();
+        eventHandlerRule.expectConversion(EVENT_NAME, USER_ID);
+
+        if (!countDownLatch.await(MAX_DURATION_MS / 2, TimeUnit.MILLISECONDS)) {
+            fail("Exceeded timeout waiting for notification.");
+        }
+    }
+
     @Test
     public void testFlushOnMismatchRevision() throws Exception {
         CountDownLatch countDownLatch = new CountDownLatch(2);
@@ -177,6 +199,31 @@ public void testFlushOnMismatchProjectId() throws Exception {
         }
     }
 
+    @Test
+    public void testStopAndStart() throws Exception {
+        CountDownLatch countDownLatch = new CountDownLatch(2);
+        setEventProcessor(logEvent -> {
+            eventHandlerRule.dispatchEvent(logEvent);
+            countDownLatch.countDown();
+        });
+
+        UserEvent userEvent = buildConversionEvent(EVENT_NAME);
+        eventProcessor.process(userEvent);
+        eventHandlerRule.expectConversion(EVENT_NAME, USER_ID);
+
+        eventProcessor.close();
+
+        eventProcessor.process(userEvent);
+        eventHandlerRule.expectConversion(EVENT_NAME, USER_ID);
+
+        eventProcessor.start();
+
+        eventProcessor.close();
+        if (!countDownLatch.await(MAX_DURATION_MS * 3, TimeUnit.MILLISECONDS)) {
+            fail("Exceeded timeout waiting for notification.");
+        }
+    }
+
     @Test
     public void testNotificationCenter() throws Exception {
         CountDownLatch countDownLatch = new CountDownLatch(1);