Merge pull request #412 from splitio/development

[7.3.4] Development into master

Author: sanzmauro

.rubocop.yml

@@ -10,16 +10,18 @@ Metrics/MethodLength:
   Max: 20
   Exclude:
     - lib/splitclient-rb/telemetry/memory/memory_synchronizer.rb
+    - lib/splitclient-rb/engine/sync_manager.rb
 
 Metrics/ClassLength:
   Max: 150
   Exclude:
     - lib/splitclient-rb/telemetry/memory/memory_synchronizer.rb
 
 Metrics/CyclomaticComplexity:
-  Max: 8
+  Max: 11
 
 Metrics/ParameterLists:
+  Max: 8
   Exclude:
     - lib/splitclient-rb/engine/sync_manager.rb
 
@@ -32,6 +34,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:

CHANGES.txt

@@ -1,6 +1,10 @@
 CHANGES
 
-7.3.3 (Jan 28, 2021)
+7.3.4 (Feb 21, 2022)
+- Updated streaming events architecture with a new queue logic.
+- Fixed redis integration Pipelining command deprecation warning.
+
+7.3.3 (Jan 28, 2022)
 - Fixed edge cases where the sdk lost streaming connection.
 - Updated default auth service url to https://auth.split.io/api/v2/auth
 - Updated dependencies:

lib/splitclient-rb/cache/adapters/redis_adapter.rb

@@ -160,7 +160,7 @@ def inc(key, inc = 1)
         end
 
         def pipelined
-          @redis.pipelined do
+          @redis.pipelined do |pipeline|
             yield
           end
         end

lib/splitclient-rb/constants.rb

@@ -12,4 +12,5 @@ class SplitIoClient::Constants
   PUSH_SUBSYSTEM_DOWN = 'PUSH_SUBSYSTEM_DOWN'
   PUSH_SUBSYSTEM_READY = 'PUSH_SUBSYSTEM_READY'
   PUSH_SUBSYSTEM_OFF = 'PUSH_SUBSYSTEM_OFF'
+  PUSH_FORCED_STOP = 'PUSH_FORCED_STOP'
 end

lib/splitclient-rb/engine/sync_manager.rb

@@ -6,35 +6,23 @@ 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,
+                     status_queue)
         @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
+        @status_queue = status_queue
+        @sse_connected = Concurrent::AtomicBoolean.new(false)
       end
 
       def start
@@ -55,6 +43,7 @@ def start_thread
 
           if @config.streaming_enabled
             @config.logger.debug('Starting Straming mode ...')
+            start_push_status_monitor
             connected = @push_manager.start_sse
           end
 
@@ -66,27 +55,6 @@ def start_thread
         end
       end
 
-      def process_action(action)
-        case action
-        when Constants::PUSH_CONNECTED
-          process_connected
-        when Constants::PUSH_RETRYABLE_ERROR
-          process_disconnect(true)
-        when Constants::PUSH_NONRETRYABLE_ERROR
-          process_disconnect(false)
-        when Constants::PUSH_SUBSYSTEM_DOWN
-          process_subsystem_down
-        when Constants::PUSH_SUBSYSTEM_READY
-          process_subsystem_ready
-        when Constants::PUSH_SUBSYSTEM_OFF
-          process_push_shutdown
-        else
-          @config.logger.debug('Incorrect action type.')
-        end
-      rescue StandardError => e
-        @config.logger.error("process_action error: #{e.inspect}")
-      end
-
       def process_subsystem_ready
         @synchronizer.stop_periodic_fetch
         @synchronizer.sync_all
@@ -124,6 +92,19 @@ def process_connected
         @config.logger.error("process_connected error: #{e.inspect}")
       end
 
+      def process_forced_stop
+        unless @sse_connected.value
+          @config.logger.debug('Streaming already disconnected.')
+          return
+        end
+
+        @sse_connected.make_false
+        @synchronizer.start_periodic_fetch
+        record_telemetry(Telemetry::Domain::Constants::SYNC_MODE, SYNC_MODE_POLLING)
+      rescue StandardError => e
+        @config.logger.error("process_connected error: #{e.inspect}")
+      end
+
       def process_disconnect(reconnect)
         unless @sse_connected.value
           @config.logger.debug('Streaming already disconnected.')
@@ -147,6 +128,40 @@ def process_disconnect(reconnect)
       def record_telemetry(type, data)
         @telemetry_runtime_producer.record_streaming_event(type, data)
       end
+
+      def start_push_status_monitor
+        @config.threads[:push_status_handler] = Thread.new do
+          @config.logger.debug('Starting push status handler ...') if @config.debug_enabled
+          incoming_push_status_handler
+        end
+      end
+
+      def incoming_push_status_handler
+        while (status = @status_queue.pop)
+          @config.logger.debug("Push status handler dequeue #{status}") if @config.debug_enabled
+
+          case status
+          when Constants::PUSH_CONNECTED
+            process_connected
+          when Constants::PUSH_RETRYABLE_ERROR
+            process_disconnect(true)
+          when Constants::PUSH_FORCED_STOP
+            process_forced_stop
+          when Constants::PUSH_NONRETRYABLE_ERROR
+            process_disconnect(false)
+          when Constants::PUSH_SUBSYSTEM_DOWN
+            process_subsystem_down
+          when Constants::PUSH_SUBSYSTEM_READY
+            process_subsystem_ready
+          when Constants::PUSH_SUBSYSTEM_OFF
+            process_push_shutdown
+          else
+            @config.logger.debug('Incorrect push status type.')
+          end
+        end
+      rescue StandardError => e
+        @config.logger.error("Push status handler error: #{e.inspect}")
+      end
     end
   end
 end

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,56 @@ 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 = Engine::Synchronizer.new(repositories, @api_key, @config, params)
+    end
+
+    def build_streaming_components
+      @push_status_queue = Queue.new
+      splits_worker = SSE::Workers::SplitsWorker.new(@synchronizer, @config, @splits_repository)
+      segments_worker = SSE::Workers::SegmentsWorker.new(@synchronizer, @config, @segments_repository)
+      notification_manager_keeper = SSE::NotificationManagerKeeper.new(@config, @runtime_producer, @push_status_queue)
+      notification_processor = 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, @push_status_queue)
+      @sse_handler = SSE::SSEHandler.new(@config, splits_worker, segments_worker, sse_client)
+      @push_manager = Engine::PushManager.new(@config, @sse_handler, @api_key, @runtime_producer)
+    end
+
+    def build_sync_manager
+      @sync_manager = Engine::SyncManager.new(@config, @synchronizer, @runtime_producer, @telemetry_synchronizer, @status_manager, @sse_handler, @push_manager, @push_status_queue)
+    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 = 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 = Engine::Common::ImpressionCounter.new
+      @impressions_manager = Engine::Common::ImpressionManager.new(@config, @impressions_repository, @impression_counter, @runtime_producer)
+    end
   end
 end