use mutex and resource to wait for flush interval or be signalled when there is something to process

thomaszurkan-optimizely · thomaszurkan-optimizely · commit af7fb6f70edb · 2019-12-12T11:36:51.000-08:00
diff --git a/lib/optimizely/event/batch_event_processor.rb b/lib/optimizely/event/batch_event_processor.rb
@@ -71,13 +71,20 @@ def start!
       end
       @flushing_interval_deadline = Helpers::DateTimeUtils.create_timestamp + @flush_interval
       @logger.log(Logger::INFO, 'Starting scheduler.')
+      if @wait_mutex.nil?
+        @wait_mutex = Mutex.new
+        @resource = ConditionVariable.new
+      end
       @thread = Thread.new { run_queue }
       @started = true
       @stopped = false
     end
 
     def flush
       @event_queue << FLUSH_SIGNAL
+      @wait_mutex.synchronize {
+        @resource.signal
+      }
     end
 
     def process(user_event)
@@ -94,6 +101,9 @@ def process(user_event)
 
       begin
         @event_queue.push(user_event, true)
+        @wait_mutex.synchronize {
+          @resource.signal
+        }
       rescue => e
         @logger.log(Logger::WARN, 'Payload not accepted by the queue: ' + e.message)
         return
@@ -105,6 +115,9 @@ def stop!
 
       @logger.log(Logger::INFO, 'Stopping scheduler.')
       @event_queue << SHUTDOWN_SIGNAL
+      @wait_mutex.synchronize {
+        @resource.signal
+      }
       @thread.join(DEFAULT_TIMEOUT_INTERVAL)
       @started = false
       @stopped = true
@@ -144,15 +157,13 @@ def run_queue
 
         break unless process_queue
 
-        # what is the current interval to flush
-        interval = (@flushing_interval_deadline - Helpers::DateTimeUtils.create_timestamp)
-
-        interval /= 10.0
-
-        # convert to seconds from milliseconds
-        interval *= 0.001
-
-        sleep interval if interval.positive?
+        # what is the current interval to flush in seconds
+        interval = (@flushing_interval_deadline - Helpers::DateTimeUtils.create_timestamp) * 0.001
+        if interval.positive?
+          @wait_mutex.synchronize {
+            @resource.wait(@wait_mutex, interval)
+          }
+        end
       end
     rescue SignalException
       @logger.log(Logger::ERROR, 'Interrupted while processing buffer.')
diff --git a/spec/event/batch_event_processor_spec.rb b/spec/event/batch_event_processor_spec.rb
@@ -344,7 +344,7 @@
     )
 
     user_event = Optimizely::UserEventFactory.create_conversion_event(project_config, event, 'test_user', nil, nil)
-    11.times do
+    12.times do
       @event_processor.process(user_event)
     end
 

Original file line number	Diff line number	Diff line change
`@@ -344,7 +344,7 @@`
`344`	`344`	`)`
`345`	`345`
`346`	`346`	`user_event = Optimizely::UserEventFactory.create_conversion_event(project_config, event, 'test_user', nil, nil)`
`347`		`- 11.times do`
	`347`	`+ 12.times do`
`348`	`348`	`@event_processor.process(user_event)`
`349`	`349`	`end`
`350`	`350`