Changed hash to array to allow multiple feature flag records

Author: chillaq

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

@@ -71,46 +71,70 @@ def add_or_update(feature_name, key)
           end
         end
 
+        def clear_cache
+          uniques = @cache.clone
+          keys_size = @keys_size
+          @cache.clear
+          @keys_size = 0
+
+          [uniques, keys_size]
+        end
+
         def send_bulk_data
           @semaphore.synchronize do
             return if @cache.empty?
 
-            uniques = @cache.clone
-            keys_size = @keys_size
-            @cache.clear
-            @keys_size = 0
-
+            uniques, keys_size = clear_cache
             if keys_size <= @max_bulk_size
               @sender_adapter.record_uniques_key(uniques)
               return
-            end
-
-            bulks = []
-            uniques.each do |unique|
-              bulks += check_keys_and_split_to_bulks(unique)
-            end
 
-            bulks.each do |b|
-              @sender_adapter.record_uniques_key(b)
             end
+            bulks = flatten_bulks(uniques)
+            bulks_to_post = group_bulks_by_max_size(bulks)
+            @sender_adapter.record_uniques_key(bulks_to_post)
           end
         rescue StandardError => e
           @config.log_found_exception(__method__.to_s, e)
         end
 
-        def check_keys_and_split_to_bulks(unique)
-          unique_updated = []
-          unique.each do |_, value|
-            if value.size > @max_bulk_size
-              sub_bulks = SplitIoClient::Utilities.split_bulk_to_send(value, value.size / @max_bulk_size)
-              sub_bulks.each do |sub_bulk|
-                unique_updated.add({ key: sub_bulk })
-              end
-              break
+        def group_bulks_by_max_size(bulks)
+          current_size = 0
+          bulks_to_post = Concurrent::Hash.new
+          bulks.each do |bulk|
+            key, value = bulk.first
+            if (value.size + current_size) > @max_bulk_size
+              @sender_adapter.record_uniques_key(bulks_to_post)
+              bulks_to_post = Concurrent::Hash.new
+              current_size = 0
+            end
+            bulks_to_post[key] = value
+            current_size += value.size
+          end
+
+          bulks_to_post
+        end
+
+        def flatten_bulks(uniques)
+          bulks = []
+          uniques.each_key do |unique_key|
+            bulks += check_keys_and_split_to_bulks(uniques[unique_key], unique_key)
+          end
 
+          bulks
+        end
+
+        def check_keys_and_split_to_bulks(value, key)
+          unique_updated = []
+          if value.size > @max_bulk_size
+            sub_bulks = SplitIoClient::Utilities.split_bulk_to_send(value, @max_bulk_size)
+            sub_bulks.each do |sub_bulk|
+              unique_updated << { key => sub_bulk.to_set }
             end
-            unique_updated.add({ key: value })
+            return unique_updated
+
           end
+          unique_updated << { key => value }
 
           unique_updated
         end

lib/splitclient-rb/utilitites.rb

@@ -38,16 +38,12 @@ def randomize_interval(interval)
       interval * random_factor
     end
 
-    def split_bulk_to_send(hash, divisions)
-      count = 0
-
-      hash.each_with_object([]) do |key_value, final|
-        final[count % divisions] ||= {}
-        final[count % divisions][key_value[0]] = key_value[1]
-        count += 1
-      end
-    rescue StandardError
-      []
+    def split_bulk_to_send(items, divisions)
+      to_return = []
+      items.to_a.each_slice(divisions) {|bulk|
+        to_return.push(bulk.to_set)
+      }
+      to_return
     end
   end  
 end

spec/engine/impressions/memory_unique_keys_tracker_spec.rb

@@ -146,5 +146,26 @@
 
       cache.clear
     end
+
+    it 'track - split chunks if above limit' do
+      cache = Concurrent::Hash.new
+      config.unique_keys_bulk_size = 1000
+      tracker = subject.new(config, filter_adapter, sender_adapter_test, cache)
+
+      10.times { |i| expect(tracker.track("feature-test-#{i}", 'key_test')).to eq(true) }
+      5.times { |i| expect(tracker.track("feature-test-1", "key_test-#{i}")).to eq(true) }
+
+      tracker.instance_variable_set(:@max_bulk_size, 5)
+      tracker.send(:send_bulk_data)
+
+      result = sender_adapter_test.bulks
+      expect(result[0].size).to eq(1)
+      expect(result[1].size).to eq(1)
+      expect(result[1]["feature-test-1"].size).to eq(5)      
+      expect(result[2].size).to eq(5)
+      expect(result[3].size).to eq(4)
+
+      cache.clear
+    end
   end
 end

spec/splitclient/utilities_spec.rb

@@ -19,37 +19,14 @@
   end
 
   it 'split bulk of data - split equally' do
-    hash = {}
+    items = Set['feature', 'feature-1', 'feature-2', 'feature-3', 'feature-4', 'feature-5', 'feature-6']
 
-    i = 1
-    while i <= 6
-      hash["mauro-#{i}"] = Set.new(['feature', 'feature-1'])
-      i += 1
-    end
-
-    result = SplitIoClient::Utilities.split_bulk_to_send(hash, 3)
+    result = SplitIoClient::Utilities.split_bulk_to_send(items, 3)
 
+    puts result
     expect(result.size).to eq 3
-    expect(result[0].size).to eq 2
-    expect(result[1].size).to eq 2
-    expect(result[2].size).to eq 2
-  end
-
-  it 'split bulk of data - split in 4 bulks' do
-    hash = {}
-
-    i = 1
-    while i <= 6
-      hash["mauro-#{i}"] = 'feature-test'
-      i += 1
-    end
-
-    result = SplitIoClient::Utilities.split_bulk_to_send(hash, 4)
-
-    expect(result.size).to eq 4
-    expect(result[0].size).to eq 2
-    expect(result[1].size).to eq 2
+    expect(result[0].size).to eq 3
+    expect(result[1].size).to eq 3
     expect(result[2].size).to eq 1
-    expect(result[3].size).to eq 1
   end
 end