chore: Create FDv2 streaming data source (#349)

Author: jsonbailey

launchdarkly-server-sdk.gemspec

@@ -43,7 +43,7 @@ Gem::Specification.new do |spec|
 
   spec.add_runtime_dependency "benchmark", "~> 0.1", ">= 0.1.1"
   spec.add_runtime_dependency "concurrent-ruby", "~> 1.1"
-  spec.add_runtime_dependency "ld-eventsource", "2.2.6"
+  spec.add_runtime_dependency "ld-eventsource", "2.4.0"
   spec.add_runtime_dependency "observer", "~> 0.1.2"
   spec.add_runtime_dependency "openssl", ">= 3.1.2", "< 5.0"
   spec.add_runtime_dependency "semantic", "~> 1.6"

lib/ldclient-rb.rb

@@ -8,6 +8,7 @@ module LaunchDarkly
 # Public APIs - these define the main interfaces users interact with
 require "ldclient-rb/config"
 require "ldclient-rb/context"
+require "ldclient-rb/data_system"
 require "ldclient-rb/flags_state"
 require "ldclient-rb/integrations"
 require "ldclient-rb/interfaces"

lib/ldclient-rb/config.rb

@@ -45,7 +45,7 @@ class Config
     # @option opts [Hash] :application See {#application}
     # @option opts [String] :payload_filter_key See {#payload_filter_key}
     # @option opts [Boolean] :omit_anonymous_contexts See {#omit_anonymous_contexts}
-    # @option opts [DataSystemConfig] :datasystem_config See {#datasystem_config}
+    # @option opts [DataSystemConfig] :data_system_config See {#data_system_config}
     # @option hooks [Array<Interfaces::Hooks::Hook]
     # @option plugins [Array<Interfaces::Plugins::Plugin]
     #
@@ -84,7 +84,7 @@ def initialize(opts = {})
       @hooks = (opts[:hooks] || []).keep_if { |hook| hook.is_a? Interfaces::Hooks::Hook }
       @plugins = (opts[:plugins] || []).keep_if { |plugin| plugin.is_a? Interfaces::Plugins::Plugin }
       @omit_anonymous_contexts = opts.has_key?(:omit_anonymous_contexts) && opts[:omit_anonymous_contexts]
-      @datasystem_config = opts[:datasystem_config]
+      @data_system_config = opts[:data_system_config]
       @data_source_update_sink = nil
       @instance_id = nil
     end
@@ -440,7 +440,7 @@ def diagnostic_opt_out?
     #
     # @return [DataSystemConfig, nil]
     #
-    attr_reader :datasystem_config
+    attr_reader :data_system_config
 
 
     #

lib/ldclient-rb/data_system.rb

@@ -3,6 +3,7 @@
 require 'ldclient-rb/interfaces/data_system'
 require 'ldclient-rb/config'
 require 'ldclient-rb/impl/data_system/polling'
+require 'ldclient-rb/impl/data_system/streaming'
 
 module LaunchDarkly
   #
@@ -132,9 +133,8 @@ def self.fdv1_fallback_ds_builder
     # @return [Proc] A proc that takes (sdk_key, config) and returns a streaming data source
     #
     def self.streaming_ds_builder
-      # TODO(fdv2): Implement streaming data source builder
-      lambda do |_sdk_key, _config|
-        raise NotImplementedError, "Streaming data source not yet implemented for FDv2"
+      lambda do |sdk_key, config|
+        LaunchDarkly::Impl::DataSystem::StreamingDataSourceBuilder.new(sdk_key, config).build
       end
     end
 

lib/ldclient-rb/impl/data_system/polling.rb

@@ -93,8 +93,6 @@ def fetch(ss)
         #
         def sync(ss)
           @logger.info { "[LDClient] Starting PollingDataSourceV2 synchronizer" }
-          @stop.reset
-          @interrupt_event.reset
 
           until @stop.set?
             result = @requester.fetch(ss.selector)
@@ -440,7 +438,7 @@ def self.polling_payload_to_changeset(data)
 
           next unless event[:event]
 
-          case event[:event]
+          case event[:event].to_sym
           when LaunchDarkly::Interfaces::DataSystem::EventName::SERVER_INTENT
             begin
               server_intent = LaunchDarkly::Interfaces::DataSystem::ServerIntent.from_h(event[:data])

lib/ldclient-rb/impl/data_system/protocolv2.rb

@@ -39,7 +39,7 @@ def initialize(version:, kind:, key:)
           #
           # Returns the event name.
           #
-          # @return [String]
+          # @return [Symbol]
           #
           def name
             LaunchDarkly::Interfaces::DataSystem::EventName::DELETE_OBJECT
@@ -111,7 +111,7 @@ def initialize(version:, kind:, key:, object:)
           #
           # Returns the event name.
           #
-          # @return [String]
+          # @return [Symbol]
           #
           def name
             LaunchDarkly::Interfaces::DataSystem::EventName::PUT_OBJECT