Merge branch 'mnoman/AddBatchEP' into mnoman/log_event_notification

# Conflicts:
#	optimizely/event/event_processor.py

Author: mariamjamal94

optimizely/event/event_processor.py

@@ -18,29 +18,27 @@
 from datetime import timedelta
 from six.moves import queue
 
-from .user_event import UserEvent
-from .event_factory import EventFactory
 from optimizely import logger as _logging
 from optimizely.event_dispatcher import EventDispatcher as default_event_dispatcher
 from optimizely.helpers import enums
 from optimizely.helpers import validator
+from .user_event import UserEvent
+from .event_factory import EventFactory
 
 ABC = abc.ABCMeta('ABC', (object,), {'__slots__': ()})
 
 
-class EventProcessor(ABC):
-  """ Class encapsulating event_processor functionality. Override with your own processor
-  providing process method. """
+class BaseEventProcessor(ABC):
+  """ Class encapsulating event processing. Override with your own implementation. """
 
   @abc.abstractmethod
   def process(user_event):
     pass
 
 
-class BatchEventProcessor(EventProcessor):
+class BatchEventProcessor(BaseEventProcessor):
   """
-  BatchEventProcessor is a batched implementation of the EventProcessor.
-
+  BatchEventProcessor is a batched implementation of the BaseEventProcessor.
   The BatchEventProcessor maintains a single consumer thread that pulls events off of
   the blocking queue and buffers them for either a configured batch size or for a
   maximum duration before the resulting LogEvent is sent to the EventDispatcher.
@@ -55,22 +53,23 @@ class BatchEventProcessor(EventProcessor):
   LOCK = threading.Lock()
 
   def __init__(self,
-                event_dispatcher,
-                logger,
-                start_on_init=False,
-                event_queue=None,
-                batch_size=None,
-                flush_interval=None,
-                timeout_interval=None,
-                notification_center=None):
+               event_dispatcher,
+               logger,
+               start_on_init=False,
+               event_queue=None,
+               batch_size=None,
+               flush_interval=None,
+               timeout_interval=None,
+               notification_center=None):
     """ EventProcessor init method to configure event batching.
+
     Args:
       event_dispatcher: Provides a dispatch_event method which if given a URL and params sends a request to it.
       logger: Provides a log method to log messages. By default nothing would be logged.
       start_on_init: Optional boolean param which starts the consumer thread if set to True.
-                     By default thread does not start unless 'start' method is called.
+                     Default value is False.
       event_queue: Optional component which accumulates the events until dispacthed.
-      batch_size: Optional param which defines the upper limit of the number of events in event_queue after which
+      batch_size: Optional param which defines the upper limit on the number of events in event_queue after which
                   the event_queue will be flushed.
       flush_interval: Optional floating point number representing time interval in seconds after which event_queue will
                       be flushed.
@@ -89,9 +88,8 @@ def __init__(self,
     self.timeout_interval = timedelta(seconds=timeout_interval) \
                               if self._validate_intantiation_props(timeout_interval, 'timeout_interval') \
                               else self._DEFAULT_TIMEOUT_INTERVAL
-
     self.notification_center = notification_center
-    self._disposed = False
+
     self._is_started = False
     self._current_batch = list()
 
@@ -100,13 +98,22 @@ def __init__(self,
 
   @property
   def is_started(self):
+    """ Property to check if consumer thread is alive or not. """
     return self._is_started
 
-  @property
-  def disposed(self):
-    return self._disposed
-
   def _validate_intantiation_props(self, prop, prop_name):
+    """ Method to determine if instantiation properties like batch_size, flush_interval
+    and timeout_interval are valid.
+
+    Args:
+      prop: Property value that needs to be validated.
+      prop_name: Property name.
+
+    Returns:
+      False if property value is None or less than 1 or not a finite number.
+      False if property name is batch_size and value is a floating point number.
+      True otherwise.
+    """
     if (prop_name == 'batch_size' and not isinstance(prop, int)) or prop is None or prop < 1 or \
       not validator.is_finite_number(prop):
       self.logger.info('Using default value for {}.'.format(prop_name))
@@ -115,13 +122,22 @@ def _validate_intantiation_props(self, prop, prop_name):
     return True
 
   def _get_time(self, _time=None):
+    """ Method to return rounded off time as integer in seconds. If _time is None, uses current time.
+
+    Args:
+      _time: time in seconds that needs to be rounded off.
+
+    Returns:
+      Integer time in seconds.
+    """
     if _time is None:
       return int(round(time.time()))
 
     return int(round(_time))
 
   def start(self):
-    if self.is_started and not self.disposed:
+    """ Starts the batch processing thread to batch events. """
+    if self.is_started:
       self.logger.warning('Service already started')
       return
 
@@ -133,7 +149,9 @@ def start(self):
     self._is_started = True
 
   def _run(self):
-    """ Scheduler method that periodically flushes events queue. """
+    """ Triggered as part of the thread which batches events or flushes event_queue and sleeps
+    periodically if queue is empty.
+    """
     try:
       while True:
         if self._get_time() > self.flushing_interval_deadline:
@@ -143,7 +161,6 @@ def _run(self):
           item = self.event_queue.get(True, 0.05)
 
         except queue.Empty:
-          self.logger.debug('Empty queue, sleeping for 50ms.')
           time.sleep(0.05)
           continue