Add cache for impressions with count

lib/splitclient-rb.rb

@@ -54,6 +54,7 @@
 require 'splitclient-rb/engine/api/segments'
 require 'splitclient-rb/engine/api/splits'
 require 'splitclient-rb/engine/api/events'
+require 'splitclient-rb/engine/common/impressions_counter'
 require 'splitclient-rb/engine/common/impressions_manager'
 require 'splitclient-rb/engine/parser/condition'
 require 'splitclient-rb/engine/parser/partition'

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

@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+require 'concurrent'
+
+module SplitIoClient
+  module Engine
+    module Common
+      TIME_INTERVAL_MS = 3600 * 1000
+
+      class ImpressionCounter
+        DEFAULT_AMOUNT = 1
+
+        def initialize
+          @cache = Concurrent::Hash.new
+        end
+
+        def inc(split_name, time_frame)
+          key = make_key(split_name, time_frame)
+
+          current_amount = @cache[key]
+          @cache[key] = current_amount.nil? ? DEFAULT_AMOUNT : (current_amount + DEFAULT_AMOUNT)
+        end
+
+        def pop_all
+          to_return = Concurrent::Hash.new
+
+          @cache.each do |key, value|
+            to_return[key] = value
+          end
+          @cache.clear
+
+          to_return
+        end
+
+        def truncate_time_frame(timestamp_ms)
+          timestamp_ms - (timestamp_ms % TIME_INTERVAL_MS)
+        end
+
+        def make_key(split_name, time_frame)
+          "#{split_name}::#{truncate_time_frame(time_frame)}"
+        end
+      end
+    end
+  end
+end

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

@@ -9,15 +9,18 @@ def initialize(config, impressions_repository)
           @impressions_repository = impressions_repository
           @impression_router = SplitIoClient::ImpressionRouter.new(@config)
           @impression_observer = SplitIoClient::Observers::ImpressionObserver.new
+          @impression_counter = SplitIoClient::Engine::Common::ImpressionCounter.new
         end
 
         # added param time for test
         def build_impression(matching_key, bucketing_key, split_name, treatment, params = {})
           impression_data = impression_data(matching_key, bucketing_key, split_name, treatment, params[:time])
 
-          impression_data[:pt] = @impression_observer.test_and_set(impression_data) if add_pt?
+          impression_data[:pt] = @impression_observer.test_and_set(impression_data) unless redis?
 
-          { m: metadata, i: impression_data, attributes: params[:attributes] }
+          return impression_optimized(split_name, impression_data, params[:attributes]) if optimized? && !redis?
+
+          impression(impression_data, params[:attributes])
         rescue StandardError => error
           @config.log_found_exception(__method__.to_s, error)
         end
@@ -57,13 +60,27 @@ def applied_rule(label)
           @config.labels_enabled ? label : nil
         end
 
-        def add_pt?
-          @config.impressions_adapter.class.to_s != 'SplitIoClient::Cache::Adapters::RedisAdapter'
-        end
-
         def optimized?
           @config.impressions_mode == :optimized
         end
+
+        def impression_optimized(split_name, impression_data, attributes)
+          @impression_counter.inc(split_name, impression_data[:m])
+
+          impression(impression_data, attributes) if should_queue_impression?(impression_data)
+        end
+
+        def should_queue_impression?(impression)
+          impression[:pt].nil? || (impression[:pt] < ((Time.now.to_f * 1000.0).to_i - Common::TIME_INTERVAL_MS))
+        end
+
+        def impression(impression_data, attributes)
+          { m: metadata, i: impression_data, attributes: attributes }
+        end
+
+        def redis?
+          @config.impressions_adapter.class.to_s == 'SplitIoClient::Cache::Adapters::RedisAdapter'
+        end
       end
     end
   end

spec/engine/common/impression_counter_spec.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+describe SplitIoClient::Engine::Common::ImpressionCounter do
+  subject { SplitIoClient::Engine::Common::ImpressionCounter }
+
+  before do
+    @counter = subject.new
+  end
+
+  it 'truncate time frame' do
+    expect(@counter.truncate_time_frame(make_timestamp('2020-09-02 10:53:12'))).to eq(make_timestamp('2020-09-02 10:00:00'))
+    expect(@counter.truncate_time_frame(make_timestamp('2020-09-02 10:00:00'))).to eq(make_timestamp('2020-09-02 10:00:00'))
+    expect(@counter.truncate_time_frame(make_timestamp('2020-09-02 10:53:00'))).to eq(make_timestamp('2020-09-02 10:00:00'))
+    expect(@counter.truncate_time_frame(make_timestamp('2020-09-02 10:00:12'))).to eq(make_timestamp('2020-09-02 10:00:00'))
+    expect(@counter.truncate_time_frame(make_timestamp('1970-01-01 00:00:00'))).to eq(make_timestamp('1970-01-01'))
+  end
+
+  it 'make key' do
+    target = make_timestamp('2020-09-02 09:00:00')
+
+    expect(@counter.make_key('feature_test', make_timestamp('2020-09-02 09:40:11'))).to eq("feature_test::#{target}")
+    expect(@counter.make_key('feature_test', make_timestamp('2020-09-02 09:25:00'))).to eq("feature_test::#{target}")
+    expect(@counter.make_key(nil, make_timestamp('2020-09-02 09:25:00'))).to eq("::#{target}")
+    expect(@counter.make_key(nil, 0)).to eq('::0')
+  end
+
+  it 'basic usage' do
+    @counter.inc('feature1', make_timestamp('2020-09-02 09:15:11'))
+    @counter.inc('feature1', make_timestamp('2020-09-02 09:20:11'))
+    @counter.inc('feature1', make_timestamp('2020-09-02 09:50:11'))
+    @counter.inc('feature2', make_timestamp('2020-09-02 09:50:11'))
+    @counter.inc('feature2', make_timestamp('2020-09-02 09:55:11'))
+    @counter.inc('feature1', make_timestamp('2020-09-02 10:50:11'))
+
+    result = @counter.pop_all
+
+    expect(result["feature1::#{make_timestamp('2020-09-02 09:00:00')}"]).to eq(3)
+    expect(result["feature2::#{make_timestamp('2020-09-02 09:00:00')}"]).to eq(2)
+    expect(result["feature1::#{make_timestamp('2020-09-02 10:00:00')}"]).to eq(1)
+
+    result = @counter.pop_all
+
+    expect(result.size).to eq(0)
+  end
+
+  def make_timestamp(time)
+    (Time.parse(time).to_f * 1000.0).to_i
+  end
+end