Add LogEvent notification. (#318)

* Add explicit LogEvent notification.
* Deprecate LogEvent from "activate" and "track" notifications.

Author: Mike Davis

core-api/src/main/java/com/optimizely/ab/Optimizely.java

@@ -963,6 +963,13 @@ public int addUpdateConfigNotificationHandler(NotificationHandler<UpdateConfigNo
         return addNotificationHandler(UpdateConfigNotification.class, handler);
     }
 
+    /**
+     * Convenience method for adding LogEvent Notification Handlers
+     */
+    public int addLogEventNotificationHandler(NotificationHandler<LogEvent> handler) {
+        return addNotificationHandler(LogEvent.class, handler);
+    }
+
     /**
      * Convenience method for adding NotificationHandlers
      */
@@ -1097,11 +1104,6 @@ public Optimizely build() {
                 decisionService = new DecisionService(bucketer, errorHandler, userProfileService);
             }
 
-            // For backwards compatibility
-            if (eventProcessor == null) {
-                eventProcessor = new ForwardingEventProcessor(eventHandler);
-            }
-
             if (projectConfig == null && datafile != null && !datafile.isEmpty()) {
                 try {
                     projectConfig = new DatafileProjectConfig.Builder().withDatafile(datafile).build();
@@ -1125,6 +1127,11 @@ public Optimizely build() {
                 notificationCenter = new NotificationCenter();
             }
 
+            // For backwards compatibility
+            if (eventProcessor == null) {
+                eventProcessor = new ForwardingEventProcessor(eventHandler, notificationCenter);
+            }
+
             return new Optimizely(eventHandler, eventProcessor, errorHandler, decisionService, userProfileService, projectConfigManager, notificationCenter);
         }
     }

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

@@ -20,6 +20,7 @@
 import com.optimizely.ab.event.internal.EventFactory;
 import com.optimizely.ab.event.internal.UserEvent;
 import com.optimizely.ab.internal.PropertyUtils;
+import com.optimizely.ab.notification.NotificationCenter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -54,15 +55,17 @@ public class BatchEventProcessor implements EventProcessor, AutoCloseable {
     private final int batchSize;
     private final long flushInterval;
     private final ExecutorService executor;
+    private final NotificationCenter notificationCenter;
 
     private Future<?> future;
     private boolean isStarted = false;
 
-    private BatchEventProcessor(BlockingQueue<Object> eventQueue, EventHandler eventHandler, Integer batchSize, Long flushInterval, ExecutorService executor) {
+    private BatchEventProcessor(BlockingQueue<Object> eventQueue, EventHandler eventHandler, Integer batchSize, Long flushInterval, ExecutorService executor, NotificationCenter notificationCenter) {
         this.eventHandler = eventHandler;
         this.eventQueue = eventQueue;
         this.batchSize = batchSize == null ? PropertyUtils.getInteger(CONFIG_BATCH_SIZE, DEFAULT_BATCH_SIZE) : batchSize;
         this.flushInterval = flushInterval == null ? PropertyUtils.getLong(CONFIG_BATCH_INTERVAL, DEFAULT_BATCH_INTERVAL) : flushInterval;
+        this.notificationCenter = notificationCenter;
 
         if (executor == null) {
             final ThreadFactory threadFactory = Executors.defaultThreadFactory();
@@ -196,6 +199,10 @@ private void flush() {
 
             LogEvent logEvent = EventFactory.createLogEvent(currentBatch);
 
+            if (notificationCenter != null) {
+                notificationCenter.send(logEvent);
+            }
+
             try {
                 eventHandler.dispatchEvent(logEvent);
             } catch (Exception e) {
@@ -215,6 +222,7 @@ public static class Builder {
         private Integer batchSize = null;
         private Long flushInterval = null;
         private ExecutorService executor = null;
+        private NotificationCenter notificationCenter = null;
 
         public Builder withEventHandler(EventHandler eventHandler) {
             this.eventHandler = eventHandler;
@@ -241,8 +249,13 @@ public Builder withExecutor(ExecutorService executor) {
             return this;
         }
 
+        public Builder withNotificationCenter(NotificationCenter notificationCenter) {
+            this.notificationCenter = notificationCenter;
+            return this;
+        }
+
         public BatchEventProcessor build() {
-            return new BatchEventProcessor(eventQueue, eventHandler, batchSize, flushInterval, executor);
+            return new BatchEventProcessor(eventQueue, eventHandler, batchSize, flushInterval, executor, notificationCenter);
         }
     }
 

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

@@ -18,6 +18,7 @@
 
 import com.optimizely.ab.event.internal.EventFactory;
 import com.optimizely.ab.event.internal.UserEvent;
+import com.optimizely.ab.notification.NotificationCenter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -30,15 +31,21 @@ public class ForwardingEventProcessor implements EventProcessor {
     private static final Logger logger = LoggerFactory.getLogger(ForwardingEventProcessor.class);
 
     private final EventHandler eventHandler;
+    private final NotificationCenter notificationCenter;
 
-    public ForwardingEventProcessor(EventHandler eventHandler) {
+    public ForwardingEventProcessor(EventHandler eventHandler, NotificationCenter notificationCenter) {
         this.eventHandler = eventHandler;
+        this.notificationCenter = notificationCenter;
     }
 
     @Override
     public void process(UserEvent userEvent) {
         LogEvent logEvent = EventFactory.createLogEvent(userEvent);
 
+        if (notificationCenter != null) {
+            notificationCenter.send(logEvent);
+        }
+
         try {
             eventHandler.dispatchEvent(logEvent);
         } catch(Exception e) {

core-api/src/main/java/com/optimizely/ab/notification/ActivateNotification.java

@@ -74,6 +74,12 @@ public Variation getVariation() {
         return variation;
     }
 
+    /**
+     * This interface is deprecated since this is no longer a one-to-one mapping.
+     * Please use a {@link NotificationHandler} explicitly for LogEvent messages.
+     * {@link com.optimizely.ab.Optimizely#addLogEventNotificationHandler(NotificationHandler)}
+     */
+    @Deprecated
     public LogEvent getEvent() {
         return event;
     }

core-api/src/main/java/com/optimizely/ab/notification/NotificationCenter.java

@@ -17,6 +17,7 @@
 package com.optimizely.ab.notification;
 
 import com.optimizely.ab.OptimizelyRuntimeException;
+import com.optimizely.ab.event.LogEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -94,6 +95,7 @@ public NotificationCenter() {
         validManagers.put(TrackNotification.class, new NotificationManager<TrackNotification>(counter));
         validManagers.put(DecisionNotification.class, new NotificationManager<DecisionNotification>(counter));
         validManagers.put(UpdateConfigNotification.class, new NotificationManager<UpdateConfigNotification>(counter));
+        validManagers.put(LogEvent.class, new NotificationManager<LogEvent>(counter));
 
         notifierMap = Collections.unmodifiableMap(validManagers);
     }

core-api/src/main/java/com/optimizely/ab/notification/TrackNotification.java

@@ -68,6 +68,12 @@ public String getUserId() {
         return eventTags;
     }
 
+    /**
+     * This interface is deprecated since this is no longer a one-to-one mapping.
+     * Please use a {@link NotificationHandler} explicitly for LogEvent messages.
+     * {@link com.optimizely.ab.Optimizely#addLogEventNotificationHandler(NotificationHandler)}
+     */
+    @Deprecated
     public LogEvent getEvent() {
         return event;
     }

core-api/src/test/java/com/optimizely/ab/OptimizelyRule.java

@@ -87,7 +87,8 @@ public void before() {
 
     public void after() {
         if (optimizely == null) {
-            return;
+            // Build so we can shut everything down.
+            build();
         }
 
         // Blocks and waits for graceful shutdown.

core-api/src/test/java/com/optimizely/ab/OptimizelyTest.java

@@ -4186,6 +4186,16 @@ public void testAddUpdateConfigNotificationHandler() {
         assertTrue(manager.remove(notificationId));
     }
 
+    @Test
+    public void testAddLogEventNotificationHandler() {
+        Optimizely optimizely = optimizelyBuilder.withConfigManager(() -> null).build();
+        NotificationManager<LogEvent> manager = optimizely.getNotificationCenter()
+            .getNotificationManager(LogEvent.class);
+
+        int notificationId = optimizely.addLogEventNotificationHandler(message -> {});
+        assertTrue(manager.remove(notificationId));
+    }
+
     //======== Helper methods ========//
 
     private Experiment createUnknownExperiment() {

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

@@ -19,6 +19,7 @@
 import com.optimizely.ab.EventHandlerRule;
 import com.optimizely.ab.config.ProjectConfig;
 import com.optimizely.ab.event.internal.*;
+import com.optimizely.ab.notification.NotificationCenter;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -52,13 +53,15 @@ public class BatchEventProcessorTest {
 
     private BlockingQueue<Object> eventQueue;
     private BatchEventProcessor eventProcessor;
+    private NotificationCenter notificationCenter;
 
     @Before
     public void setUp() throws Exception {
         when(projectConfig.getRevision()).thenReturn("1");
         when(projectConfig.getProjectId()).thenReturn("X");
 
         eventQueue = new ArrayBlockingQueue<>(100);
+        notificationCenter = new NotificationCenter();
     }
 
     @After
@@ -173,12 +176,28 @@ public void testFlushOnMismatchProjectId() throws Exception {
         }
     }
 
+    @Test
+    public void testNotificationCenter() throws Exception {
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        notificationCenter.addNotificationHandler(LogEvent.class, x -> countDownLatch.countDown());
+        setEventProcessor(logEvent -> {});
+
+        UserEvent userEvent = buildConversionEvent(EVENT_NAME);
+        eventProcessor.process(userEvent);
+        eventProcessor.close();
+
+        if (!countDownLatch.await(MAX_DURATION_MS * 3, TimeUnit.MILLISECONDS)) {
+            fail("Exceeded timeout waiting for notification.");
+        }
+    }
+
     private void setEventProcessor(EventHandler eventHandler) {
         eventProcessor = BatchEventProcessor.builder()
             .withEventQueue(eventQueue)
             .withBatchSize(MAX_BATCH_SIZE)
             .withFlushInterval(MAX_DURATION_MS)
             .withEventHandler(eventHandler)
+            .withNotificationCenter(notificationCenter)
             .build();
     }
 

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

@@ -18,6 +18,7 @@
 
 import com.optimizely.ab.config.ProjectConfig;
 import com.optimizely.ab.event.internal.*;
+import com.optimizely.ab.notification.NotificationCenter;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -38,6 +39,7 @@ public class ForwardingEventProcessorTest {
 
     private ForwardingEventProcessor eventProcessor;
     private AtomicBoolean atomicBoolean = new AtomicBoolean();
+    private NotificationCenter notificationCenter = new NotificationCenter();
 
     @Mock
     private ProjectConfig projectConfig;
@@ -50,16 +52,26 @@ public void setUp() throws Exception {
             assertEquals(logEvent.getRequestMethod(), LogEvent.RequestMethod.POST);
             assertEquals(logEvent.getEndpointUrl(), EventFactory.EVENT_ENDPOINT);
             atomicBoolean.set(true);
-        });
+        }, notificationCenter);
     }
 
     @Test
-    public void testAddHandler() {
+    public void testEventHandler() {
         UserEvent userEvent = buildConversionEvent(EVENT_NAME);
         eventProcessor.process(userEvent);
         assertTrue(atomicBoolean.get());
     }
 
+    @Test
+    public void testNotifications() {
+        AtomicBoolean notifcationTriggered = new AtomicBoolean();
+        notificationCenter.addNotificationHandler(LogEvent.class, x -> notifcationTriggered.set(true));
+        UserEvent userEvent = buildConversionEvent(EVENT_NAME);
+        eventProcessor.process(userEvent);
+        assertTrue(atomicBoolean.get());
+        assertTrue(notifcationTriggered.get());
+    }
+
     private ConversionEvent buildConversionEvent(String eventName) {
         return UserEventFactory.createConversionEvent(projectConfig, USER_ID, EVENT_ID, eventName,
             Collections.emptyMap(), Collections.emptyMap());