[PART 2] Tests - Implement deduping logic in redis mode

lib/splitclient-rb.rb

@@ -15,6 +15,7 @@
 require 'splitclient-rb/cache/filter/filter_adapter'
 require 'splitclient-rb/cache/hashers/impression_hasher'
 require 'splitclient-rb/cache/observers/impression_observer'
+require 'splitclient-rb/cache/observers/noop_impression_observer'
 require 'splitclient-rb/cache/repositories/repository'
 require 'splitclient-rb/cache/repositories/segments_repository'
 require 'splitclient-rb/cache/repositories/splits_repository'
@@ -56,6 +57,7 @@
 require 'splitclient-rb/engine/api/telemetry_api'
 require 'splitclient-rb/engine/common/impressions_counter'
 require 'splitclient-rb/engine/common/impressions_manager'
+require 'splitclient-rb/engine/common/noop_impressions_counter'
 require 'splitclient-rb/engine/parser/condition'
 require 'splitclient-rb/engine/parser/partition'
 require 'splitclient-rb/engine/parser/evaluator'
@@ -83,6 +85,7 @@
 require 'splitclient-rb/engine/matchers/equal_to_matcher'
 require 'splitclient-rb/engine/matchers/matches_string_matcher'
 require 'splitclient-rb/engine/evaluator/splitter'
+require 'splitclient-rb/engine/impressions/noop_unique_keys_tracker'
 require 'splitclient-rb/engine/impressions/unique_keys_tracker'
 require 'splitclient-rb/engine/metrics/binary_search_latency_tracker'
 require 'splitclient-rb/engine/models/split'

lib/splitclient-rb/cache/observers/noop_impression_observer.rb

@@ -0,0 +1,10 @@
+module SplitIoClient
+  module Observers
+    class NoopImpressionObserver
+      def test_and_set(impression)
+        # no-op
+      end
+    end
+  end
+end
+

lib/splitclient-rb/cache/senders/impressions_adapter/memory_sender.rb

@@ -46,7 +46,7 @@ def uniques_formatter(uniques)
         end
 
         def impressions_count_formatter(counts)
-          return if counts.empty?
+          return if counts.nil? || counts.empty?
 
           formated_counts = {pf: []}
 

lib/splitclient-rb/cache/senders/impressions_adapter/redis_sender.rb

@@ -10,7 +10,11 @@ def initialize(config)
         end
 
         def record_uniques_key(uniques)
-          # TODO: implementation
+          formatted = uniques_formatter(uniques)
+
+          @adapter.add_to_queue(unique_keys_key, formatted) unless formatted.nil?
+        rescue StandardError => e
+          @config.log_found_exception(__method__.to_s, e)
         end
 
         def record_impressions_count(impressions_count)
@@ -28,6 +32,27 @@ def record_impressions_count(impressions_count)
         def impressions_count_key
           "#{@config.redis_namespace}.impressions.count"
         end
+
+        def unique_keys_key
+          "#{@config.redis_namespace}.uniquekeys"
+        end
+
+        def uniques_formatter(uniques)
+          return if uniques.empty?
+
+          to_return = []
+          uniques.each do |key, value|
+            to_return << {
+              f: key,
+              k: value.to_a
+            }
+          end
+
+          to_return
+        rescue StandardError => error
+          @config.log_found_exception(__method__.to_s, error)
+          nil
+        end
       end
     end
   end

lib/splitclient-rb/cache/senders/impressions_count_sender.rb

@@ -20,7 +20,6 @@ def impressions_count_thread
           @config.threads[:impressions_count_sender] = Thread.new do
             begin
               @config.logger.info('Starting impressions count service')
-
               loop do
                 sleep(@config.counter_refresh_rate)
 

lib/splitclient-rb/engine/common/noop_impressions_counter.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+require 'concurrent'
+
+module SplitIoClient
+  module Engine
+    module Common
+      class NoopImpressionCounter
+        def inc(split_name, time_frame)
+          # no-op
+        end
+
+        def pop_all
+          # no-op
+        end
+
+        def make_key(split_name, time_frame)
+          # no-op
+        end
+
+        def self.truncate_time_frame(timestamp_ms)
+          # no-op
+        end
+      end
+    end
+  end
+end

lib/splitclient-rb/engine/impressions/noop_unique_keys_tracker.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+module SplitIoClient
+  module Engine
+    module Impressions
+      class NoopUniqueKeysTracker
+        def call
+          # no-op
+        end
+
+        def track(feature_name, key)
+          # no-op
+        end
+      end
+    end
+  end
+end

lib/splitclient-rb/engine/impressions/unique_keys_tracker.rb

@@ -49,6 +49,7 @@ def send_bulk_data_thread
           end
         rescue SplitIoClient::SDKShutdownException
           send_bulk_data
+          @config.logger.info('Posting unique keys due to shutdown')
         end
 
         def clear_filter_thread

lib/splitclient-rb/engine/synchronizer.rb

@@ -26,6 +26,7 @@ def initialize(
         @impression_counter = params[:imp_counter]
         @telemetry_synchronizer = params[:telemetry_synchronizer]
         @impressions_sender_adapter = params[:impressions_sender_adapter]
+        @unique_keys_tracker = params[:unique_keys_tracker]
       end
 
       def sync_all(asynchronous = true)
@@ -42,9 +43,10 @@ def sync_all(asynchronous = true)
 
       def start_periodic_data_recording
         impressions_sender
-        events_sender
         impressions_count_sender
+        events_sender
         start_telemetry_sync_task
+        start_unique_keys_tracker_task
       end
 
       def start_periodic_fetch
@@ -169,7 +171,7 @@ def attempt_splits_sync(target_cn, fetch_options, max_retries, retry_delay_secon
 
       # Starts thread which loops constantly and sends impressions to the Split API
       def impressions_sender
-        ImpressionsSender.new(@impressions_repository, @config, @impressions_api).call
+        ImpressionsSender.new(@impressions_repository, @config, @impressions_api).call unless @config.impressions_mode == :none
       end
 
       # Starts thread which loops constantly and sends events to the Split API
@@ -179,13 +181,17 @@ def events_sender
 
       # Starts thread which loops constantly and sends impressions count to the Split API
       def impressions_count_sender
-        ImpressionsCountSender.new(@config, @impression_counter, @impressions_sender_adapter).call
+        ImpressionsCountSender.new(@config, @impression_counter, @impressions_sender_adapter).call unless @config.impressions_mode == :debug
       end
 
       def start_telemetry_sync_task
         Telemetry::SyncTask.new(@config, @telemetry_synchronizer).call
       end
 
+      def start_unique_keys_tracker_task
+        @unique_keys_tracker.call
+      end
+
       def sync_result(success, remaining_attempts, segment_names = nil)
         { success: success, remaining_attempts: remaining_attempts, segment_names: segment_names }
       end

lib/splitclient-rb/split_config.rb

@@ -297,12 +297,9 @@ def initialize(opts = {})
     attr_accessor :counter_refresh_rate
 
     def self.default_counter_refresh_rate(adapter)
-      case adapter
-      when :redis
-        300 # Send bulk impressions count - Refresh rate: 5 min.
-      else
-        1800 # Send bulk impressions count - Refresh rate: 30 min.
-      end
+      return 300 if adapter == :redis # Send bulk impressions count - Refresh rate: 5 min.
+
+      1800 # Send bulk impressions count - Refresh rate: 30 min.
     end
 
     def self.default_on_demand_fetch_retry_delay_seconds

lib/splitclient-rb/split_factory.rb

@@ -38,9 +38,10 @@ def initialize(api_key, config_hash = {})
 
       build_telemetry_components
       build_repositories
+      build_telemetry_synchronizer
+      build_impressions_sender_adapter
       build_unique_keys_tracker
       build_impressions_components
-      build_telemetry_synchronizer
 
       @status_manager = Engine::StatusManager.new(@config)
 
@@ -174,7 +175,8 @@ def build_synchronizer
         imp_counter: @impression_counter,
         telemetry_synchronizer: @telemetry_synchronizer,
         impressions_sender_adapter: @impressions_sender_adapter,
-        impressions_api: @impressions_api
+        impressions_api: @impressions_api,
+        unique_keys_tracker: @unique_keys_tracker
       }
 
       @synchronizer = Engine::Synchronizer.new(repositories, @config, params)
@@ -209,20 +211,46 @@ def build_telemetry_synchronizer
     end
 
     def build_unique_keys_tracker
+      if @config.impressions_mode != :none
+        @unique_keys_tracker = Engine::Impressions::NoopUniqueKeysTracker.new
+        return
+      end
+
       bf = BloomFilter::Native.new(size: 95_850_584, hashes: 2)
       filter_adapter = Cache::Filter::FilterAdapter.new(@config, bf)
-      @impressions_api = Api::Impressions.new(@api_key, @config, @runtime_producer)
-      @impressions_sender_adapter = Cache::Senders::ImpressionsSenderAdapter.new(config, @telemetry_api, @impressions_api)
       cache = Concurrent::Hash.new
       @unique_keys_tracker = Engine::Impressions::UniqueKeysTracker.new(@config, filter_adapter, @impressions_sender_adapter, cache)
     end
 
+    def build_impressions_observer
+      if (@config.cache_adapter == :redis && @config.impressions_mode != :optimized) ||
+         (@config.cache_adapter == :memory && @config.impressions_mode == :none)
+        @impression_observer = Observers::NoopImpressionObserver.new
+      else
+        @impression_observer = Observers::ImpressionObserver.new
+      end
+    end
+
+    def build_impression_counter
+      case @config.impressions_mode
+      when :debug
+        @impression_counter = Engine::Common::NoopImpressionCounter.new
+      else
+        @impression_counter = Engine::Common::ImpressionCounter.new
+      end
+    end
+
+    def build_impressions_sender_adapter
+      @impressions_api = Api::Impressions.new(@api_key, @config, @runtime_producer)
+      @impressions_sender_adapter = Cache::Senders::ImpressionsSenderAdapter.new(@config, @telemetry_api, @impressions_api)
+    end
+
     def build_impressions_components
-      @impression_counter = Engine::Common::ImpressionCounter.new
-      impression_observer = Observers::ImpressionObserver.new
-      impression_router = ImpressionRouter.new(@config)
+      build_impressions_observer
+      build_impression_counter
 
-      @impressions_manager = Engine::Common::ImpressionManager.new(@config, @impressions_repository, @impression_counter, @runtime_producer, impression_observer, @unique_keys_tracker, impression_router)
+      impression_router = ImpressionRouter.new(@config)
+      @impressions_manager = Engine::Common::ImpressionManager.new(@config, @impressions_repository, @impression_counter, @runtime_producer, @impression_observer, @unique_keys_tracker, impression_router)
     end
   end
 end

spec/allocations/splitclient-rb/clients/split_client_spec.rb

@@ -17,7 +17,12 @@
   let(:runtime_producer) { SplitIoClient::Telemetry::RuntimeProducer.new(config) }
   let(:api_key) { 'SplitClient-key' }
   let(:telemetry_api) { SplitIoClient::Api::TelemetryApi.new(config, api_key, runtime_producer) }
-  let(:sender_adapter) { SplitIoClient::Cache::Senders::UniqueKeysSenderAdapter.new(config, telemetry_api) }
+  let(:impressions_api) { SplitIoClient::Api::Impressions.new(api_key, config, runtime_producer) }
+  let(:sender_adapter) do
+    SplitIoClient::Cache::Senders::ImpressionsSenderAdapter.new(config,
+                                                                telemetry_api,
+                                                                impressions_api)
+  end
   let(:unique_keys_tracker) do
     SplitIoClient::Engine::Impressions::UniqueKeysTracker.new(config,
                                                               filter_adapter,

spec/cache/repositories/impressions_repository_spec.rb

@@ -112,7 +112,8 @@
       filter_adapter = SplitIoClient::Cache::Filter::FilterAdapter.new(config, bf)
       api_key = 'ImpressionsRepository-memory-key'
       telemetry_api = SplitIoClient::Api::TelemetryApi.new(config, api_key, runtime_producer)
-      sender_adapter = SplitIoClient::Cache::Senders::UniqueKeysSenderAdapter.new(config, telemetry_api)
+      impressions_api = SplitIoClient::Api::Impressions.new(api_key, config, runtime_producer)
+      sender_adapter = SplitIoClient::Cache::Senders::ImpressionsSenderAdapter.new(config, telemetry_api, impressions_api)
 
       SplitIoClient::Engine::Impressions::UniqueKeysTracker.new(config,
                                                                 filter_adapter,
@@ -170,7 +171,8 @@
       filter_adapter = SplitIoClient::Cache::Filter::FilterAdapter.new(config, bf)
       api_key = 'ImpressionsRepository-key'
       telemetry_api = SplitIoClient::Api::TelemetryApi.new(config, api_key, runtime_producer)
-      sender_adapter = SplitIoClient::Cache::Senders::UniqueKeysSenderAdapter.new(config, telemetry_api)
+      impressions_api = SplitIoClient::Api::Impressions.new(api_key, config, runtime_producer)
+      sender_adapter = SplitIoClient::Cache::Senders::ImpressionsSenderAdapter.new(config, telemetry_api, impressions_api)
 
       SplitIoClient::Engine::Impressions::UniqueKeysTracker.new(config,
                                                                 filter_adapter,