Fme 12889 events task

lib/splitclient-rb.rb

@@ -67,6 +67,7 @@
 require 'splitclient-rb/engine/common/impressions_manager'
 require 'splitclient-rb/engine/common/noop_impressions_counter'
 require 'splitclient-rb/engine/events/events_manager_config.rb'
+require 'splitclient-rb/engine/events/events_task.rb'
 require 'splitclient-rb/engine/parser/condition'
 require 'splitclient-rb/engine/parser/partition'
 require 'splitclient-rb/engine/parser/evaluator'
@@ -117,6 +118,7 @@
 require 'splitclient-rb/engine/models/sdk_event_type.rb'
 require 'splitclient-rb/engine/models/sdk_event.rb'
 require 'splitclient-rb/engine/models/sdk_internal_event.rb'
+require 'splitclient-rb/engine/models/sdk_internal_event_notification.rb'
 require 'splitclient-rb/engine/auth_api_client'
 require 'splitclient-rb/engine/back_off'
 require 'splitclient-rb/engine/fallback_treatment_calculator.rb'

lib/splitclient-rb/engine/events/events_manager_config.rb

@@ -1,87 +1,96 @@
 # frozen_string_literal: true
 
-module SplitIoClient::Engine::Events
-    class EventsManagerConfig
+module SplitIoClient
+  module Engine
+    module Events
+      class EventsManagerConfig
         attr_accessor :require_all, :prerequisites, :require_any, :suppressed_by, :execution_limits, :evaluation_order
 
         def initialize
-            @require_all = get_require_all
-            @prerequisites = get_prerequisites
-            @require_any = get_require_any
-            @suppressed_by = get_suppressed_by
-            @execution_limits = get_execution_limits
-            @evaluation_order = get_sorted_events
+          @require_all = construct_require_all
+          @prerequisites = construct_prerequisites
+          @require_any = construct_require_any
+          @suppressed_by = construct_suppressed_by
+          @execution_limits = construct_execution_limits
+          @evaluation_order = construct_sorted_events
         end
-                        
+
         private
 
-        def get_require_all
-            return  {
-                        SplitIoClient::Engine::Models::SdkEvent::SDK_READY => Set.new([SplitIoClient::Engine::Models::SdkInternalEvent::SDK_READY])
-                    }
+        def construct_require_all
+          {
+            SplitIoClient::Engine::Models::SdkEvent::SDK_READY => Set.new([SplitIoClient::Engine::Models::SdkInternalEvent::SDK_READY])
+          }
         end
 
-        def get_prerequisites
-            return  {
-                        SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE => Set.new([SplitIoClient::Engine::Models::SdkEvent::SDK_READY])
-                    }
+        def construct_prerequisites
+          {
+            SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE => Set.new([SplitIoClient::Engine::Models::SdkEvent::SDK_READY])
+          }
         end
 
-        def get_require_any
-            return  {
-                        SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE => Set.new([SplitIoClient::Engine::Models::SdkInternalEvent::FLAG_KILLED_NOTIFICATION, SplitIoClient::Engine::Models::SdkInternalEvent::FLAGS_UPDATED, 
-                                                SplitIoClient::Engine::Models::SdkInternalEvent::RB_SEGMENTS_UPDATED, SplitIoClient::Engine::Models::SdkInternalEvent::SEGMENTS_UPDATED])
-                    }
+        def construct_require_any
+          {
+            SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE => Set.new(
+              [
+                SplitIoClient::Engine::Models::SdkInternalEvent::FLAG_KILLED_NOTIFICATION,
+                SplitIoClient::Engine::Models::SdkInternalEvent::FLAGS_UPDATED,
+                SplitIoClient::Engine::Models::SdkInternalEvent::RB_SEGMENTS_UPDATED,
+                SplitIoClient::Engine::Models::SdkInternalEvent::SEGMENTS_UPDATED
+              ]
+            )
+          }
         end
 
-        def get_suppressed_by
-            return  {}
+        def construct_suppressed_by
+          {}
         end
 
-        def get_execution_limits
-            return  {
-                        SplitIoClient::Engine::Models::SdkEvent::SDK_READY => 1,
-                        SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE => -1
-                    }
+        def construct_execution_limits
+          {
+            SplitIoClient::Engine::Models::SdkEvent::SDK_READY => 1,
+            SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE => -1
+          }
         end
 
-        def get_sorted_events
-            sorted_events = []
-            for sdk_event in [SplitIoClient::Engine::Models::SdkEvent::SDK_READY, SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE]
-                sorted_events = dfs_recursive(sdk_event, sorted_events)
-            end
+        def construct_sorted_events
+          sorted_events = []
+          [SplitIoClient::Engine::Models::SdkEvent::SDK_READY, SplitIoClient::Engine::Models::SdkEvent::SDK_UPDATE].each do |sdk_event|
+            sorted_events = dfs_recursive(sdk_event, sorted_events)
+          end
 
-            return sorted_events
+          sorted_events
         end
 
         def dfs_recursive(sdk_event, added)
-            return added if added.include?(sdk_event)
+          return added if added.include?(sdk_event)
 
-            get_dependencies(sdk_event).each do |dependent_event|
-                added = dfs_recursive(dependent_event, added)
-            end
+          get_dependencies(sdk_event).each do |dependent_event|
+            added = dfs_recursive(dependent_event, added)
+          end
 
-            added.push(sdk_event)
-            return added
+          added.push(sdk_event)
+
+          added
         end
 
         def get_dependencies(sdk_event)
-            dependencies = Set.new
-            @prerequisites.each do |prerequisites_event_name, prerequisites_event_value|
-                if prerequisites_event_name == sdk_event
-                    for prereq_event in prerequisites_event_value
-                        dependencies.add(prereq_event)
-                    end
-                end
-            end
+          dependencies = Set.new
+          @prerequisites.each do |prerequisites_event_name, prerequisites_event_value|
+            next unless prerequisites_event_name == sdk_event
 
-            @suppressed_by.each do |suppressed_event_name, suppressed_event_value|
-                if suppressed_event_value.include?(sdk_event)
-                    dependencies.add(suppressed_event_name)
-                end
+            prerequisites_event_value.each do |prereq_event|
+              dependencies.add(prereq_event)
             end
+          end
+
+          @suppressed_by.each do |suppressed_event_name, suppressed_event_value|
+            dependencies.add(suppressed_event_name) if suppressed_event_value.include?(sdk_event)
+          end
 
-            return dependencies
+          dependencies
         end
+      end
     end
-end
+  end
+end

lib/splitclient-rb/engine/events/events_task.rb

@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module SplitIoClient
+  module Engine
+    module Events
+      class EventsTask
+        attr_accessor :running
+
+        def initialize(notify_internal_events, internal_events_queue, config)
+          @notify_internal_events = notify_internal_events
+          @internal_events_queue = internal_events_queue
+          @config = config
+          @running = false
+        end
+
+        def start
+          return if @running
+
+          @config.logger.info('Starting Internal Events Task.') if @config.debug_enabled
+          @running = true
+          @config.threads[:internal_events_task] = Thread.new do
+            worker_thread
+          end
+        end
+
+        def stop
+          return unless @running
+
+          @config.logger.info('Stopping Internal Events Task.') if @config.debug_enabled
+          @running = false
+        end
+
+        private
+
+        def worker_thread
+          while (event = @internal_events_queue.pop)
+            break unless @running
+
+            @config.logger.info("Processing sdk internal event: #{event.internal_event}") if @config.debug_enabled
+            begin
+              @notify_internal_events.call(event.internal_event, event.metadata)
+            rescue StandardError => e
+              @config.log_found_exception(__method__.to_s, e)
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/splitclient-rb/engine/models/sdk_internal_event_notification.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: false
+
+module SplitIoClient
+  module Engine::Models
+    class SdkInternalEventNotification
+      attr_reader :internal_event, :metadata
+
+      def initialize(internal_event, metadata)
+        @internal_event = internal_event
+        @metadata = metadata
+      end
+    end
+  end
+end

spec/engine/events/events_task_spec.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe SplitIoClient::Engine::Events::EventsTask do
+  subject { SplitIoClient::Engine::Events::EventsTask }
+    let(:internal_event) { nil }
+    let(:metadata) { nil }
+
+    it 'test_task_running' do
+        queue = Queue.new
+        config = SplitIoClient::SplitConfig.new(logger: Logger.new(StringIO.new))
+        task = subject.new(method(:call_back), queue, config)
+        task.start
+        expect(task.running).to be(true)
+
+        queue.push(SplitIoClient::Engine::Models::SdkInternalEventNotification.new(SplitIoClient::Engine::Models::SdkInternalEvent::FLAGS_UPDATED, SplitIoClient::Engine::Models::EventsMetadata.new(SplitIoClient::Engine::Models::SdkEventType::FLAG_UPDATE)))
+        sleep 0.5
+        expect(@internal_event).to be(SplitIoClient::Engine::Models::SdkInternalEvent::FLAGS_UPDATED)
+        expect(@metadata.type).to be(SplitIoClient::Engine::Models::SdkEventType::FLAG_UPDATE)
+
+        @internal_event = nil
+        @metadata = nil
+        queue.push(SplitIoClient::Engine::Models::SdkInternalEventNotification.new(SplitIoClient::Engine::Models::SdkInternalEvent::RB_SEGMENTS_UPDATED, SplitIoClient::Engine::Models::EventsMetadata.new(SplitIoClient::Engine::Models::SdkEventType::SEGMENTS_UPDATE)))
+        sleep 0.5
+        expect(@internal_event).to be(SplitIoClient::Engine::Models::SdkInternalEvent::RB_SEGMENTS_UPDATED)
+        expect(@metadata.type).to be(SplitIoClient::Engine::Models::SdkEventType::SEGMENTS_UPDATE)
+
+        task.stop
+        sleep 0.2
+        expect(task.running).to be(false)
+
+    end
+
+    def call_back(internal_event, metadata)
+        @internal_event = internal_event
+        @metadata = metadata
+    end
+
+end