removed streaming events logic.

Author: sanzmauro

.rubocop.yml

@@ -20,6 +20,7 @@ Metrics/CyclomaticComplexity:
   Max: 8
 
 Metrics/ParameterLists:
+  Max: 8
   Exclude:
     - lib/splitclient-rb/engine/sync_manager.rb
 
@@ -32,6 +33,7 @@ Metrics/LineLength:
     - spec/engine/auth_api_client_spec.rb
     - spec/telemetry/synchronizer_spec.rb
     - spec/splitclient/split_config_spec.rb
+    - spec/engine/push_manager_spec.rb
 
 Style/BracesAroundHashParameters:
   Exclude:

lib/splitclient-rb/engine/sync_manager.rb

@@ -6,35 +6,21 @@ class SyncManager
       SYNC_MODE_STREAMING = 0
       SYNC_MODE_POLLING = 1
 
-      def initialize(
-        repositories,
-        api_key,
-        config,
-        synchronizer,
-        telemetry_runtime_producer,
-        telemetry_synchronizer,
-        status_manager
-      )
-        @synchronizer = synchronizer
-        notification_manager_keeper = SSE::NotificationManagerKeeper.new(config, telemetry_runtime_producer) do |manager|
-          manager.on_action { |action| process_action(action) }
-        end
-        @sse_handler = SSE::SSEHandler.new(
-          { config: config, api_key: api_key },
-          @synchronizer,
-          repositories,
-          notification_manager_keeper,
-          telemetry_runtime_producer
-        ) do |handler|
-          handler.on_action { |action| process_action(action) }
-        end
-
-        @push_manager = PushManager.new(config, @sse_handler, api_key, telemetry_runtime_producer)
-        @sse_connected = Concurrent::AtomicBoolean.new(false)
+      def initialize(config,
+                     synchronizer,
+                     telemetry_runtime_producer,
+                     telemetry_synchronizer,
+                     status_manager,
+                     sse_handler,
+                     push_manager)
         @config = config
+        @synchronizer = synchronizer
         @telemetry_runtime_producer = telemetry_runtime_producer
         @telemetry_synchronizer = telemetry_synchronizer
         @status_manager = status_manager
+        @sse_handler = sse_handler
+        @push_manager = push_manager
+        @sse_connected = Concurrent::AtomicBoolean.new(false)
       end
 
       def start

lib/splitclient-rb/split_factory.rb

@@ -33,15 +33,10 @@ def initialize(api_key, config_hash = {})
       register_factory
 
       build_telemetry_components
+      build_repositories
+      build_impressions_components
+      build_telemetry_synchronizer
 
-      @splits_repository = SplitsRepository.new(@config)
-      @segments_repository = SegmentsRepository.new(@config)
-      @impressions_repository = ImpressionsRepository.new(@config)
-      @events_repository = EventsRepository.new(@config, @api_key, @runtime_producer)
-      @impression_counter = SplitIoClient::Engine::Common::ImpressionCounter.new
-      @impressions_manager = SplitIoClient::Engine::Common::ImpressionManager.new(@config, @impressions_repository, @impression_counter, @runtime_producer)
-      @telemetry_api = SplitIoClient::Api::TelemetryApi.new(@config, @api_key, @runtime_producer)
-      @telemetry_synchronizer = Telemetry::Synchronizer.new(@config, @telemetry_consumers, @init_producer, repositories, @telemetry_api)
       @status_manager = Engine::StatusManager.new(@config)
 
       start!
@@ -59,18 +54,12 @@ def start!
         return
       end
 
-      split_fetcher = SplitFetcher.new(@splits_repository, @api_key, config, @runtime_producer)
-      segment_fetcher = SegmentFetcher.new(@segments_repository, @api_key, config, @runtime_producer)
-      params = {
-        split_fetcher: split_fetcher,
-        segment_fetcher: segment_fetcher,
-        imp_counter: @impression_counter,
-        telemetry_runtime_producer: @runtime_producer,
-        telemetry_synchronizer: @telemetry_synchronizer
-      }
+      build_fetchers
+      build_synchronizer
+      build_streaming_components
+      build_sync_manager
 
-      synchronizer = SplitIoClient::Engine::Synchronizer.new(repositories, @api_key, @config, params)
-      SplitIoClient::Engine::SyncManager.new(repositories, @api_key, @config, synchronizer, @runtime_producer, @telemetry_synchronizer, @status_manager).start
+      @sync_manager.start
     end
 
     def stop!
@@ -166,6 +155,55 @@ def build_telemetry_components
       @runtime_producer = Telemetry::RuntimeProducer.new(@config)
 
       @telemetry_consumers = { init: @init_consumer, evaluation: @evaluation_consumer, runtime: @runtime_consumer }
-    end    
+    end
+
+    def build_fetchers
+      @split_fetcher = SplitFetcher.new(@splits_repository, @api_key, @config, @runtime_producer)
+      @segment_fetcher = SegmentFetcher.new(@segments_repository, @api_key, @config, @runtime_producer)
+    end
+
+    def build_synchronizer
+      params = {
+        split_fetcher: @split_fetcher,
+        segment_fetcher: @segment_fetcher,
+        imp_counter: @impression_counter,
+        telemetry_runtime_producer: @runtime_producer,
+        telemetry_synchronizer: @telemetry_synchronizer
+      }
+
+      @synchronizer = SplitIoClient::Engine::Synchronizer.new(repositories, @api_key, @config, params)
+    end
+
+    def build_sync_manager
+      @sync_manager = SplitIoClient::Engine::SyncManager.new(@config, @synchronizer, @runtime_producer, @telemetry_synchronizer, @status_manager, @sse_handler, @push_manager)
+    end
+
+    def build_streaming_components      
+      splits_worker = SplitIoClient::SSE::Workers::SplitsWorker.new(@synchronizer, @config, @splits_repository)
+      segments_worker = SplitIoClient::SSE::Workers::SegmentsWorker.new(synchronizer, @config, @segments_repository)
+      notification_manager_keeper = SSE::NotificationManagerKeeper.new(@config, @runtime_producer)
+      notification_processor = SplitIoClient::SSE::NotificationProcessor.new(@config, @splits_worker, @segments_worker)
+      event_parser = SSE::EventSource::EventParser.new(config)
+      sse_client = SSE::EventSource::Client.new(@config, @api_key, @runtime_producer, event_parser, notification_manager_keeper, notification_processor)
+      @sse_handler = SSE::SSEHandler.new(@config, splits_worker, segments_worker, sse_client)
+      @push_manager = PushManager.new(@config, @sse_handler, @api_key, @runtime_producer)
+    end
+
+    def build_repositories
+      @splits_repository = SplitsRepository.new(@config)
+      @segments_repository = SegmentsRepository.new(@config)
+      @impressions_repository = ImpressionsRepository.new(@config)
+      @events_repository = EventsRepository.new(@config, @api_key, @runtime_producer)
+    end
+
+    def build_telemetry_synchronizer
+      telemetry_api = SplitIoClient::Api::TelemetryApi.new(@config, @api_key, @runtime_producer)
+      @telemetry_synchronizer = Telemetry::Synchronizer.new(@config, @telemetry_consumers, @init_producer, repositories, telemetry_api)
+    end
+
+    def build_impressions_components
+      @impression_counter = SplitIoClient::Engine::Common::ImpressionCounter.new
+      @impressions_manager = SplitIoClient::Engine::Common::ImpressionManager.new(@config, @impressions_repository, @impression_counter, @runtime_producer)
+    end
   end
 end

lib/splitclient-rb/sse/event_source/client.rb

@@ -13,26 +13,23 @@ class Client
         KEEP_ALIVE_RESPONSE = "c\r\n:keepalive\n\n\r\n".freeze
         ERROR_EVENT_TYPE = 'error'.freeze
 
-        def initialize(config, api_key, telemetry_runtime_producer, read_timeout: DEFAULT_READ_TIMEOUT)
+        def initialize(config,
+                       api_key,
+                       telemetry_runtime_producer,
+                       event_parser,
+                       notification_manager_keeper,
+                       notification_processor,
+                       read_timeout: DEFAULT_READ_TIMEOUT)
           @config = config
+          @api_key = api_key
+          @telemetry_runtime_producer = telemetry_runtime_producer
+          @event_parser = event_parser
+          @notification_manager_keeper = notification_manager_keeper
+          @notification_processor = notification_processor
           @read_timeout = read_timeout
           @connected = Concurrent::AtomicBoolean.new(false)
           @first_event = Concurrent::AtomicBoolean.new(true)
           @socket = nil
-          @event_parser = SSE::EventSource::EventParser.new(config)
-          @on = { event: ->(_) {}, action: ->(_) {} }
-          @api_key = api_key
-          @telemetry_runtime_producer = telemetry_runtime_producer
-
-          yield self if block_given?
-        end
-
-        def on_event(&action)
-          @on[:event] = action
-        end
-
-        def on_action(&action)
-          @on[:action] = action
         end
 
         def close(action = nil)
@@ -188,15 +185,15 @@ def dispatch_error(event)
         end
 
         def dispatch_event(event)
-          @config.logger.debug("Dispatching event: #{event.event_type}, #{event.channel}") if @config.debug_enabled
-          @on[:event].call(event)
+          if event.occupancy?
+            @notification_manager_keeper.handle_incoming_occupancy_event(event)
+          else
+            @notification_processor.process(event)
+          end
         end
 
         def dispatch_action(action)
-          @config.threads[:dispatch_action] = Thread.new do
-            @config.logger.debug("Dispatching action: #{action}") if @config.debug_enabled
-            @on[:action].call(action)
-          end
+          # TODO: will use status queue here.
         end
       end
     end

lib/splitclient-rb/sse/notification_manager_keeper.rb

@@ -14,10 +14,7 @@ def initialize(config, telemetry_runtime_producer)
         @publisher_available = Concurrent::AtomicBoolean.new(true)
         @publishers_pri = Concurrent::AtomicFixnum.new
         @publishers_sec = Concurrent::AtomicFixnum.new
-        @on = { action: ->(_) {} }
         @telemetry_runtime_producer = telemetry_runtime_producer
-
-        yield self if block_given?
       end
 
       def handle_incoming_occupancy_event(event)
@@ -27,14 +24,9 @@ def handle_incoming_occupancy_event(event)
           process_event_occupancy(event.channel, event.data['metrics']['publishers'])
         end
       rescue StandardError => e
-        p e
         @config.logger.error(e)
       end
 
-      def on_action(&action)
-        @on[:action] = action
-      end
-
       private
 
       def process_event_control(type)
@@ -83,7 +75,7 @@ def are_publishers_available?
 
       def dispatch_action(action)
         @config.logger.debug("Dispatching action: #{action}")
-        @on[:action].call(action)
+        # TODO: will use status queue here.
       end
     end
   end

lib/splitclient-rb/sse/sse_handler.rb

@@ -5,24 +5,14 @@ module SSE
     class SSEHandler
       attr_reader :sse_client
 
-      def initialize(metadata,
-                     synchronizer,
-                     repositories,
-                     notification_manager_keeper,
-                     telemetry_runtime_producer)
-        @config = metadata[:config]
-        @notification_manager_keeper = notification_manager_keeper
-        @splits_worker = SplitIoClient::SSE::Workers::SplitsWorker.new(synchronizer, @config, repositories[:splits])
-        @segments_worker = SplitIoClient::SSE::Workers::SegmentsWorker.new(synchronizer, @config, repositories[:segments])
-        @notification_processor = SplitIoClient::SSE::NotificationProcessor.new(@config, @splits_worker, @segments_worker)
-        @sse_client = SSE::EventSource::Client.new(@config, metadata[:api_key], telemetry_runtime_producer) do |client|
-          client.on_event { |event| handle_incoming_message(event) }
-          client.on_action { |action| process_action(action) }
-        end
-
-        @on = { action: ->(_) {} }
-
-        yield self if block_given?
+      def initialize(config,
+                     splits_worker,
+                     segments_worker,
+                     sse_client)
+        @config = config
+        @splits_worker = splits_worker
+        @segments_worker = segments_worker
+        @sse_client = sse_client
       end
 
       def start(token_jwt, channels)
@@ -50,24 +40,6 @@ def stop_workers
         @splits_worker.stop
         @segments_worker.stop
       end
-
-      def on_action(&action)
-        @on[:action] = action
-      end
-
-      private
-
-      def process_action(action)
-        @on[:action].call(action)
-      end
-
-      def handle_incoming_message(notification)
-        if notification.occupancy?
-          @notification_manager_keeper.handle_incoming_occupancy_event(notification)
-        else
-          @notification_processor.process(notification)
-        end
-      end
     end
   end
 end