Implement Event Batching (#312)

* Introduces an EventProcessor interface.
* Introduces a BatchEventProcessor (maybe should name it BatchingEventProcessor)
* Introduces an OptimizelyRule to handle lifecycle management during unit testing.

Buffering events within a queue before dispatching is an optimization that should prevent SDK implementations from exhausting resources while increasing throughput. This implementation relies on a `BlockingQueue` to buffer events received from one-to-many producers. A single consumer thread continuously polls from this queue to build a batch before emitting the batched `LogEvent`.

Note that the `Optimizely` class only knows about the `EventProcessor` and `EventHandler` so that it can gracefully close those implementations, so not strictly necessary. The Optimizely class builds UserEvents via the `UserEventFactory` and passes those events to the `EventProcessor`. The `EventProcessor` emits `LogEvents` to a `NotificationManager<LogEvent>` which is subscribed to by the `EventHandler` for eventual dispatching.

Author: Mike Davis

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

@@ -24,9 +24,7 @@
 import com.optimizely.ab.config.parser.ConfigParseException;
 import com.optimizely.ab.error.ErrorHandler;
 import com.optimizely.ab.error.NoOpErrorHandler;
-import com.optimizely.ab.event.EventHandler;
-import com.optimizely.ab.event.LogEvent;
-import com.optimizely.ab.event.NoopEventHandler;
+import com.optimizely.ab.event.*;
 import com.optimizely.ab.event.internal.*;
 import com.optimizely.ab.event.internal.payload.EventBatch;
 import com.optimizely.ab.notification.*;
@@ -78,6 +76,8 @@ public class Optimizely implements AutoCloseable {
     @VisibleForTesting
     final EventHandler eventHandler;
     @VisibleForTesting
+    final EventProcessor eventProcessor;
+    @VisibleForTesting
     final ErrorHandler errorHandler;
 
     private final ProjectConfigManager projectConfigManager;
@@ -89,13 +89,15 @@ public class Optimizely implements AutoCloseable {
     private final UserProfileService userProfileService;
 
     private Optimizely(@Nonnull EventHandler eventHandler,
+                       @Nonnull EventProcessor eventProcessor,
                        @Nonnull ErrorHandler errorHandler,
                        @Nonnull DecisionService decisionService,
                        @Nullable UserProfileService userProfileService,
                        @Nonnull ProjectConfigManager projectConfigManager,
                        @Nonnull NotificationCenter notificationCenter
     ) {
         this.eventHandler = eventHandler;
+        this.eventProcessor = eventProcessor;
         this.errorHandler = errorHandler;
         this.decisionService = decisionService;
         this.userProfileService = userProfileService;
@@ -137,6 +139,7 @@ private void tryClose(Object obj) {
      */
     @Override
     public void close() {
+        tryClose(eventProcessor);
         tryClose(eventHandler);
         tryClose(projectConfigManager);
     }
@@ -230,28 +233,25 @@ private void sendImpression(@Nonnull ProjectConfig projectConfig,
             return;
         }
 
-        logger.info("Activating user \"{}\" in experiment \"{}\".", userId, experiment.getKey());
         UserEvent userEvent = UserEventFactory.createImpressionEvent(
             projectConfig,
             experiment,
             variation,
             userId,
             filteredAttributes);
 
-        LogEvent impressionEvent = EventFactory.createLogEvent(userEvent);
-
-        try {
-            eventHandler.dispatchEvent(impressionEvent);
-        } catch (Exception e) {
-            logger.error("Unexpected exception in event dispatcher", e);
-        }
+        eventProcessor.process(userEvent);
+        logger.info("Activating user \"{}\" in experiment \"{}\".", userId, experiment.getKey());
 
         // Kept For backwards compatibility.
         // This notification is deprecated and the new DecisionNotifications
         // are sent via their respective method calls.
-        ActivateNotification activateNotification = new ActivateNotification(
-            experiment, userId, filteredAttributes, variation, impressionEvent);
-        notificationCenter.send(activateNotification);
+        if (notificationCenter.getNotificationManager(ActivateNotification.class).size() > 0) {
+            LogEvent impressionEvent = EventFactory.createLogEvent(userEvent);
+            ActivateNotification activateNotification = new ActivateNotification(
+                experiment, userId, filteredAttributes, variation, impressionEvent);
+            notificationCenter.send(activateNotification);
+        }
     }
 
     //======== track calls ========//
@@ -309,20 +309,17 @@ public void track(@Nonnull String eventName,
             copiedAttributes,
             eventTags);
 
-        // create the conversion event request parameters, then dispatch
-        LogEvent conversionEvent = EventFactory.createLogEvent(userEvent);
+        eventProcessor.process(userEvent);
         logger.info("Tracking event \"{}\" for user \"{}\".", eventName, userId);
 
-        try {
-            eventHandler.dispatchEvent(conversionEvent);
-        } catch (Exception e) {
-            logger.error("Unexpected exception in event dispatcher", e);
-        }
-
-        TrackNotification notification = new TrackNotification(eventName, userId,
-            copiedAttributes, eventTags, conversionEvent);
+        if (notificationCenter.getNotificationManager(TrackNotification.class).size() > 0) {
+            // create the conversion event request parameters, then dispatch
+            LogEvent conversionEvent = EventFactory.createLogEvent(userEvent);
+            TrackNotification notification = new TrackNotification(eventName, userId,
+                copiedAttributes, eventTags, conversionEvent);
 
-        notificationCenter.send(notification);
+            notificationCenter.send(notification);
+        }
     }
 
     //======== FeatureFlag APIs ========//
@@ -1000,6 +997,7 @@ public static class Builder {
         private DecisionService decisionService;
         private ErrorHandler errorHandler;
         private EventHandler eventHandler;
+        private EventProcessor eventProcessor;
         private ProjectConfig projectConfig;
         private ProjectConfigManager projectConfigManager;
         private UserProfileService userProfileService;
@@ -1027,6 +1025,11 @@ public Builder withEventHandler(EventHandler eventHandler) {
             return this;
         }
 
+        public Builder withEventProcessor(EventProcessor eventProcessor) {
+            this.eventProcessor = eventProcessor;
+            return this;
+        }
+
         public Builder withUserProfileService(UserProfileService userProfileService) {
             this.userProfileService = userProfileService;
             return this;
@@ -1094,6 +1097,11 @@ 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();
@@ -1117,7 +1125,7 @@ public Optimizely build() {
                 notificationCenter = new NotificationCenter();
             }
 
-            return new Optimizely(eventHandler, errorHandler, decisionService, userProfileService, projectConfigManager, notificationCenter);
+            return new Optimizely(eventHandler, eventProcessor, errorHandler, decisionService, userProfileService, projectConfigManager, notificationCenter);
         }
     }
 }