Merge pull request #549 from splitio/rbs-fetcher

Added RBS support for fetcher

Author: chillaq

lib/splitclient-rb/cache/fetchers/split_fetcher.rb

@@ -2,10 +2,11 @@ module SplitIoClient
   module Cache
     module Fetchers
       class SplitFetcher
-        attr_reader :splits_repository
+        attr_reader :splits_repository, :rule_based_segments_repository
 
-        def initialize(splits_repository, api_key, config, telemetry_runtime_producer)
+        def initialize(splits_repository, rule_based_segments_repository, api_key, config, telemetry_runtime_producer)
           @splits_repository = splits_repository
+          @rule_based_segments_repository = rule_based_segments_repository
           @api_key = api_key
           @config = config
           @semaphore = Mutex.new
@@ -23,9 +24,10 @@ def call
 
         def fetch_splits(fetch_options = { cache_control_headers: false, till: nil })
           @semaphore.synchronize do
-            data = splits_since(@splits_repository.get_change_number, fetch_options)
+            data = splits_since(@splits_repository.get_change_number, @rule_based_segments_repository.get_change_number, fetch_options)
 
-            SplitIoClient::Helpers::RepositoryHelper.update_feature_flag_repository(@splits_repository, data[:splits], data[:till], @config)
+            SplitIoClient::Helpers::RepositoryHelper.update_feature_flag_repository(@splits_repository, data[:ff][:d], data[:ff][:t], @config)
+            SplitIoClient::Helpers::RepositoryHelper.update_rule_based_segment_repository(@rule_based_segments_repository, data[:rbs][:d], data[:rbs][:t], @config)
             @splits_repository.set_segment_names(data[:segment_names])
             @config.logger.debug("segments seen(#{data[:segment_names].length}): #{data[:segment_names].to_a}") if @config.debug_enabled
 
@@ -55,8 +57,8 @@ def splits_thread
           end
         end
 
-        def splits_since(since, fetch_options = { cache_control_headers: false, till: nil })
-          splits_api.since(since, fetch_options)
+        def splits_since(since, since_rbs, fetch_options = { cache_control_headers: false, till: nil })
+          splits_api.since(since, since_rbs, fetch_options)
         end
 
         def splits_api

lib/splitclient-rb/engine/api/splits.rb

@@ -12,10 +12,10 @@ def initialize(api_key, config, telemetry_runtime_producer)
         @flag_sets_filter = @config.flag_sets_filter
       end
 
-      def since(since, fetch_options = { cache_control_headers: false, till: nil, sets: nil})
+      def since(since, since_rbs, fetch_options = { cache_control_headers: false, till: nil, sets: nil})
         start = Time.now
 
-        params = { s: SplitIoClient::Spec::FeatureFlags::SPEC_VERSION, since: since }
+        params = { s: SplitIoClient::Spec::FeatureFlags::SPEC_VERSION, since: since, rbSince: since_rbs }
         params[:sets] = @flag_sets_filter.join(",") unless @flag_sets_filter.empty?
         params[:till] = fetch_options[:till] unless fetch_options[:till].nil?
         @config.logger.debug("Fetching from splitChanges with #{params}: ")
@@ -25,11 +25,17 @@ def since(since, fetch_options = { cache_control_headers: false, till: nil, sets
           raise ApiException.new response.body, 414
         end
         if response.success?
-          result = splits_with_segment_names(response.body)
-          unless result[:splits].empty?
-            @config.split_logger.log_if_debug("#{result[:splits].length} feature flags retrieved. since=#{since}")
+          result = objects_with_segment_names(response.body)
+          
+          unless result[:ff][:d].empty?
+            @config.split_logger.log_if_debug("#{result[:ff][:d].length} feature flags retrieved. since=#{since}")
           end
-          @config.split_logger.log_if_transport("Feature flag changes response: #{result.to_s}")
+          @config.split_logger.log_if_transport("Feature flag changes response: #{result[:ff].to_s}")
+
+          unless result[:rbs][:d].empty?
+            @config.split_logger.log_if_debug("#{result[:ff][:d].length} rule based segments retrieved. since=#{since_rbs}")
+          end
+          @config.split_logger.log_if_transport("rule based segments changes response: #{result[:rbs].to_s}")
 
           bucket = BinarySearchLatencyTracker.get_bucket((Time.now - start) * 1000.0)
           @telemetry_runtime_producer.record_sync_latency(Telemetry::Domain::Constants::SPLIT_SYNC, bucket)
@@ -48,15 +54,19 @@ def since(since, fetch_options = { cache_control_headers: false, till: nil, sets
 
       private
 
-      def splits_with_segment_names(splits_json)
-        parsed_splits = JSON.parse(splits_json, symbolize_names: true)
+      def objects_with_segment_names(objects_json)
+        parsed_objects = JSON.parse(objects_json, symbolize_names: true)
 
-        parsed_splits[:segment_names] =
-          parsed_splits[:splits].each_with_object(Set.new) do |split, splits|
-            splits << Helpers::Util.segment_names_by_feature_flag(split)
+        parsed_objects[:segment_names] =
+          parsed_objects[:ff][:d].each_with_object(Set.new) do |split, splits|
+            splits << Helpers::Util.segment_names_by_object(split)
           end.flatten
-
-        parsed_splits
+        if not parsed_objects[:ff][:rbs].nil?
+          parsed_objects[:segment_names].merge parsed_objects[:ff][:rbs].each_with_object(Set.new) do |rule_based_segment, rule_based_segments|
+            rule_based_segments << Helpers::Util.segment_names_by_object(rule_based_segment)
+          end.flatten
+        end
+        parsed_objects
       end
     end
   end

lib/splitclient-rb/helpers/repository_helper.rb

@@ -25,6 +25,22 @@ def self.update_feature_flag_repository(feature_flag_repository, feature_flags,
         end
         feature_flag_repository.update(to_add, to_delete, change_number)
       end
+
+      def self.update_rule_based_segment_repository(rule_based_segment_repository, rule_based_segments, change_number, config)
+        to_add = []
+        to_delete = []
+        rule_based_segments.each do |rule_based_segment|
+          if Engine::Models::Split.archived?(rule_based_segment)
+            config.logger.debug("removing rule based segment from store(#{rule_based_segment})") if config.debug_enabled
+            to_delete.push(rule_based_segment)
+            next
+          end
+
+          config.logger.debug("storing rule based segment (#{rule_based_segment[:name]})") if config.debug_enabled
+          to_add.push(rule_based_segment)
+        end
+        rule_based_segment_repository.update(to_add, to_delete, change_number)
+      end
     end
   end
 end

lib/splitclient-rb/helpers/util.rb

@@ -3,8 +3,8 @@
 module SplitIoClient
   module Helpers
     class Util
-      def self.segment_names_by_feature_flag(feature_flag)
-        feature_flag[:conditions].each_with_object(Set.new) do |condition, names|
+      def self.segment_names_by_object(object)
+        object[:conditions].each_with_object(Set.new) do |condition, names|
           condition[:matcherGroup][:matchers].each do |matcher|
             next if matcher[:userDefinedSegmentMatcherData].nil?
 

lib/splitclient-rb/spec.rb

@@ -3,7 +3,7 @@
 module SplitIoClient
   module Spec
     class FeatureFlags
-      SPEC_VERSION = "1.1"
+      SPEC_VERSION = "1.3"
     end
   end
 end

spec/cache/fetchers/segment_fetch_spec.rb

@@ -36,10 +36,11 @@
     let(:flag_sets_repository) {SplitIoClient::Cache::Repositories::MemoryFlagSetsRepository.new([]) }
     let(:flag_set_filter) {SplitIoClient::Cache::Filter::FlagSetsFilter.new([]) }
     let(:splits_repository) { SplitIoClient::Cache::Repositories::SplitsRepository.new(config, flag_sets_repository, flag_set_filter) }
+    let(:rule_based_segments_repository) { SplitIoClient::Cache::Repositories::RuleBasedSegmentsRepository.new(config) }
     let(:telemetry_runtime_producer) { SplitIoClient::Telemetry::RuntimeProducer.new(config) }
     let(:segment_fetcher) { described_class.new(segments_repository, '', config, telemetry_runtime_producer) }
     let(:split_fetcher) do
-      SplitIoClient::Cache::Fetchers::SplitFetcher.new(splits_repository, '', config, telemetry_runtime_producer)
+      SplitIoClient::Cache::Fetchers::SplitFetcher.new(splits_repository, rule_based_segments_repository, '', config, telemetry_runtime_producer)
     end
 
     it 'fetch segments' do
@@ -72,10 +73,11 @@
     let(:flag_sets_repository) {SplitIoClient::Cache::Repositories::RedisFlagSetsRepository.new(config) }
     let(:flag_set_filter) {SplitIoClient::Cache::Filter::FlagSetsFilter.new([]) }
     let(:splits_repository) { SplitIoClient::Cache::Repositories::SplitsRepository.new(config, flag_sets_repository, flag_set_filter) }
+    let(:rule_based_segments_repository) { SplitIoClient::Cache::Repositories::RuleBasedSegmentsRepository.new(config) }
     let(:telemetry_runtime_producer) { SplitIoClient::Telemetry::RuntimeProducer.new(config) }
     let(:segment_fetcher) { described_class.new(segments_repository, '', config, telemetry_runtime_producer) }
     let(:split_fetcher) do
-      SplitIoClient::Cache::Fetchers::SplitFetcher.new(splits_repository, '', config, telemetry_runtime_producer)
+      SplitIoClient::Cache::Fetchers::SplitFetcher.new(splits_repository, rule_based_segments_repository, '', config, telemetry_runtime_producer)
     end
 
     it 'fetch segments' do

spec/cache/fetchers/split_fetch_spec.rb

@@ -11,7 +11,7 @@
   end
 
   before do
-    stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.1&since=-1')
+    stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.3&since=-1&rbSince=-1')
       .to_return(status: 200, body: active_splits_json)
   end
 
@@ -26,20 +26,23 @@
     let(:flag_sets_repository) {SplitIoClient::Cache::Repositories::MemoryFlagSetsRepository.new([]) }
     let(:flag_set_filter) {SplitIoClient::Cache::Filter::FlagSetsFilter.new([]) }
     let(:splits_repository) { SplitIoClient::Cache::Repositories::SplitsRepository.new(config, flag_sets_repository, flag_set_filter) }
+    let(:rule_based_segments_repository) { SplitIoClient::Cache::Repositories::RuleBasedSegmentsRepository.new(config) }
     let(:telemetry_runtime_producer) { SplitIoClient::Telemetry::RuntimeProducer.new(config) }
-    let(:store) { described_class.new(splits_repository, '', config, telemetry_runtime_producer) }
+    let(:store) { described_class.new(splits_repository, rule_based_segments_repository, '', config, telemetry_runtime_producer) }
 
     it 'returns splits since' do
-      splits = store.send(:splits_since, -1)
+      splits = store.send(:splits_since, -1, -1)
 
-      expect(splits[:splits].count).to eq(2)
+      expect(splits[:ff][:d].count).to eq(2)
+      expect(splits[:rbs][:d].count).to eq(1)
     end
 
     it 'fetch data in the cache' do
       store.send(:fetch_splits)
-
       expect(store.splits_repository.splits.size).to eq(2)
-      expect(store.splits_repository.get_change_number).to eq(store.send(:splits_since, -1)[:till])
+      expect(store.splits_repository.get_change_number).to eq(store.send(:splits_since, -1, -1)[:ff][:t])
+      expect(store.rule_based_segments_repository.rule_based_segment_names.size).to eq(1)
+      expect(store.rule_based_segments_repository.get_change_number).to eq(store.send(:splits_since, -1, -1)[:rbs][:t])
     end
 
     it 'refreshes splits' do
@@ -48,7 +51,7 @@
       active_split = store.splits_repository.splits['test_1_ruby']
       expect(active_split[:status]).to eq('ACTIVE')
 
-      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.1&since=1473413807667')
+      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.3&since=1473413807667&rbSince=1457726098069')
         .to_return(status: 200, body: archived_splits_json)
 
       store.send(:fetch_splits)
@@ -77,40 +80,42 @@
     let(:flag_sets_repository) {SplitIoClient::Cache::Repositories::MemoryFlagSetsRepository.new(['set_2']) }
     let(:flag_set_filter) {SplitIoClient::Cache::Filter::FlagSetsFilter.new(['set_2']) }
     let(:splits_repository) { SplitIoClient::Cache::Repositories::SplitsRepository.new(config, flag_sets_repository, flag_set_filter) }
+    let(:rule_based_segments_repository) { SplitIoClient::Cache::Repositories::RuleBasedSegmentsRepository.new(config) }
     let(:telemetry_runtime_producer) { SplitIoClient::Telemetry::RuntimeProducer.new(config) }
-    let(:store) { described_class.new(splits_repository, '', config, telemetry_runtime_producer) }
+    let(:store) { described_class.new(splits_repository, rule_based_segments_repository, '', config, telemetry_runtime_producer) }
 
     before do
-      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.1&since=-1&sets=set_2')
+      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.3&since=-1&rbSince=-1&sets=set_2')
       .to_return(status: 200, body: active_splits_json)
     end
 
     it 'returns splits since' do
-      splits = store.send(:splits_since, -1)
+      splits = store.send(:splits_since, -1, -1)
 
-      expect(splits[:splits].count).to eq(2)
+      expect(splits[:ff][:d].count).to eq(2)
     end
 
     it 'fetch data in the cache' do
       store.send(:fetch_splits)
 
       expect(store.splits_repository.splits.size).to eq(1)
-      expect(store.splits_repository.get_change_number).to eq(store.send(:splits_since, -1)[:till])
+      expect(store.splits_repository.get_change_number).to eq(store.send(:splits_since, -1, -1)[:ff][:t])
     end
 
     it 'refreshes splits' do
       store.send(:fetch_splits)
       expect(store.splits_repository.get_split('sample_feature')[:name]).to eq('sample_feature')
       expect(store.splits_repository.get_split('test_1_ruby')).to eq(nil)
 
-      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.1&since=1473413807667&sets=set_2')
+      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.3&since=1473413807667&rbSince=1457726098069&sets=set_2')
         .to_return(status: 200, body: archived_splits_json)
 
       store.send(:fetch_splits)
       expect(store.splits_repository.get_split('sample_feature')).to eq(nil)
 
       store.splits_repository.set_change_number(-1)
-      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.1&since=-1&sets=set_2')
+      store.rule_based_segments_repository.set_change_number(-1)
+      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.3&since=-1&rbSince=-1&sets=set_2')
         .to_return(status: 200, body: active_splits_json)
 
       store.send(:fetch_splits)
@@ -129,18 +134,19 @@
     let(:flag_sets_repository) {SplitIoClient::Cache::Repositories::RedisFlagSetsRepository.new(config) }
     let(:flag_set_filter) {SplitIoClient::Cache::Filter::FlagSetsFilter.new([]) }
     let(:splits_repository) { SplitIoClient::Cache::Repositories::SplitsRepository.new(config, flag_sets_repository, flag_set_filter) }
+    let(:rule_based_segments_repository) { SplitIoClient::Cache::Repositories::RuleBasedSegmentsRepository.new(config) }
     let(:telemetry_runtime_producer) { SplitIoClient::Telemetry::RuntimeProducer.new(config) }
-    let(:store) { described_class.new(splits_repository, '', config, telemetry_runtime_producer) }
+    let(:store) { described_class.new(splits_repository, rule_based_segments_repository, '', config, telemetry_runtime_producer) }
 
     it 'returns splits since' do
-      splits = store.send(:splits_since, -1)
-      expect(splits[:splits].count).to eq(2)
+      splits = store.send(:splits_since, -1, -1)
+      expect(splits[:ff][:d].count).to eq(2)
     end
 
     it 'fetch data in the cache' do
       store.send(:fetch_splits)
       expect(store.splits_repository.splits.size).to eq(2)
-      expect(store.splits_repository.get_change_number).to eq(store.send(:splits_since, -1)[:till].to_s)
+      expect(store.splits_repository.get_change_number).to eq(store.send(:splits_since, -1, -1)[:ff][:t].to_s)
     end
 
     it 'refreshes splits' do
@@ -149,7 +155,7 @@
       active_split = store.splits_repository.splits['test_1_ruby']
       expect(active_split[:status]).to eq('ACTIVE')
 
-      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.1&since=1473413807667')
+      stub_request(:get, 'https://sdk.split.io/api/splitChanges?s=1.3&since=1473413807667&rbSince=1457726098069')
         .to_return(status: 200, body: archived_splits_json)
 
       store.send(:fetch_splits)