feat: streaming support

examples/simple.rb

@@ -18,17 +18,16 @@
 # or:
 
 @unleash = Unleash::Client.new(
-  url: 'https://unleash.herokuapp.com/api',
-  custom_http_headers: { 'Authorization': '943ca9171e2c884c545c5d82417a655fb77cec970cc3b78a8ff87f4406b495d0' },
+  url: 'https://app.unleash-hosted.com/demo/api',
+  custom_http_headers: { 'Authorization': 'demo-app:dev.9fc74dd72d2b88bea5253c04240b21a54841f08d9918046ed55a06b5' },
   app_name: 'simple-test',
   instance_id: 'local-test-cli',
   refresh_interval: 2,
   metrics_interval: 2,
   retry_limit: 2
 )
 
-# feature_name = "AwesomeFeature"
-feature_name = "4343443"
+feature_name = "example-flag"
 unleash_context = Unleash::Context.new
 unleash_context.user_id = 123
 

examples/streaming.rb

@@ -0,0 +1,50 @@
+#!/usr/bin/env ruby
+
+require 'unleash'
+require 'unleash/context'
+
+puts ">> START streaming.rb"
+
+@unleash = Unleash::Client.new(
+  url: 'https://app.unleash-hosted.com/demo/api',
+  custom_http_headers: { 'Authorization': 'demo-app:dev.9fc74dd72d2b88bea5253c04240b21a54841f08d9918046ed55a06b5' },
+  app_name: 'streaming-test',
+  instance_id: 'local-streaming-cli',
+  refresh_interval: 2,
+  metrics_interval: 2,
+  retry_limit: 2,
+  experimental_mode: { type: 'streaming' },
+  timeout: 5,
+  log_level: Logger::DEBUG
+)
+
+feature_name = "example-flag"
+unleash_context = Unleash::Context.new
+unleash_context.user_id = 123
+
+puts "Waiting for client to initialize..."
+sleep 2
+
+100.times do
+  if @unleash.is_enabled?(feature_name, unleash_context)
+    puts "> #{feature_name} is enabled"
+  else
+    puts "> #{feature_name} is not enabled"
+  end
+  sleep 1
+  puts "---"
+  puts ""
+  puts ""
+end
+feature_name = "foobar"
+if @unleash.is_enabled?(feature_name, unleash_context, true)
+  puts "> #{feature_name} is enabled"
+else
+  puts "> #{feature_name} is not enabled"
+end
+
+puts "> shutting down client..."
+
+@unleash.shutdown
+
+puts ">> END streaming.rb"

lib/unleash/client.rb

@@ -2,14 +2,16 @@
 require 'unleash/toggle_fetcher'
 require 'unleash/metrics_reporter'
 require 'unleash/scheduled_executor'
+require 'unleash/streaming_client'
 require 'unleash/variant'
 require 'unleash/util/http'
+require 'ld-eventsource'
 require 'logger'
 require 'time'
 
 module Unleash
   class Client
-    attr_accessor :fetcher_scheduled_executor, :metrics_scheduled_executor
+    attr_accessor :fetcher_scheduled_executor, :metrics_scheduled_executor, :streaming_client
 
     # rubocop:disable Metrics/AbcSize
     def initialize(*opts)
@@ -21,7 +23,8 @@ def initialize(*opts)
       Unleash.engine = YggdrasilEngine.new
       Unleash.engine.register_custom_strategies(Unleash.configuration.strategies.custom_strategies)
 
-      Unleash.toggle_fetcher = Unleash::ToggleFetcher.new Unleash.engine
+      Unleash.toggle_fetcher = Unleash::ToggleFetcher.new Unleash.engine unless Unleash.configuration.streaming_mode?
+
       if Unleash.configuration.disable_client
         Unleash.logger.warn "Unleash::Client is disabled! Will only return default (or bootstrapped if available) results!"
         Unleash.logger.warn "Unleash::Client is disabled! Metrics and MetricsReporter are also disabled!"
@@ -30,7 +33,13 @@ def initialize(*opts)
       end
 
       register
-      start_toggle_fetcher
+
+      if Unleash.configuration.streaming_mode?
+        start_streaming_client
+      else
+        start_toggle_fetcher
+      end
+
       start_metrics unless Unleash.configuration.disable_metrics
     end
     # rubocop:enable Metrics/AbcSize
@@ -105,7 +114,11 @@ def shutdown
     # quick shutdown: just kill running threads
     def shutdown!
       unless Unleash.configuration.disable_client
-        self.fetcher_scheduled_executor.exit
+        if Unleash.configuration.streaming_mode?
+          self.streaming_client.stop
+        else
+          self.fetcher_scheduled_executor.exit
+        end
         self.metrics_scheduled_executor.exit unless Unleash.configuration.disable_metrics
       end
     end
@@ -140,6 +153,11 @@ def start_toggle_fetcher
       end
     end
 
+    def start_streaming_client
+      self.streaming_client = Unleash::StreamingClient.new Unleash.engine
+      self.streaming_client.start
+    end
+
     def start_metrics
       Unleash.reporter = Unleash::MetricsReporter.new
       self.metrics_scheduled_executor = Unleash::ScheduledExecutor.new(

lib/unleash/configuration.rb

@@ -22,7 +22,8 @@ class Configuration
       :log_level,
       :bootstrap_config,
       :strategies,
-      :use_delta_api
+      :use_delta_api,
+      :experimental_mode
     attr_reader :connection_id
 
     def initialize(opts = {})
@@ -67,7 +68,9 @@ def fetch_toggles_uri
       uri = nil
       ## Personal feeling but Rubocop's suggestion here is too dense to be properly readable
       # rubocop:disable Style/ConditionalAssignment
-      if self.use_delta_api
+      if streaming_mode?
+        uri = URI("#{self.url_stripped_of_slash}/client/streaming")
+      elsif self.use_delta_api || polling_with_delta?
         uri = URI("#{self.url_stripped_of_slash}/client/delta")
       else
         uri = URI("#{self.url_stripped_of_slash}/client/features")
@@ -93,6 +96,16 @@ def use_bootstrap?
       self.bootstrap_config&.valid?
     end
 
+    def streaming_mode?
+      self.experimental_mode.is_a?(Hash) && self.experimental_mode[:type] == 'streaming'
+    end
+
+    def polling_with_delta?
+      self.experimental_mode.is_a?(Hash) &&
+        self.experimental_mode[:type] == 'polling' &&
+        self.experimental_mode[:format] == 'delta'
+    end
+
     private
 
     def set_defaults
@@ -112,6 +125,7 @@ def set_defaults
       self.bootstrap_config = nil
       self.strategies       = Unleash::Strategies.new
       self.use_delta_api    = false
+      self.experimental_mode = nil
 
       self.custom_http_headers = {}
       @connection_id = SecureRandom.uuid

lib/unleash/streaming_client.rb

@@ -0,0 +1,93 @@
+require 'ld-eventsource'
+require 'json'
+
+module Unleash
+  class StreamingClient
+    attr_accessor :event_source, :toggle_engine, :running, :mutex
+
+    def initialize(toggle_engine)
+      self.toggle_engine = toggle_engine
+      self.event_source = nil
+      self.running = false
+      self.mutex = Mutex.new
+    end
+
+    def start
+      self.mutex.synchronize do
+        return if self.running || Unleash.configuration.disable_client
+
+        Unleash.logger.debug "Starting streaming from URL: #{Unleash.configuration.fetch_toggles_uri}"
+
+        headers = (Unleash.configuration.http_headers || {}).dup
+
+        self.event_source = SSE::Client.new(
+          Unleash.configuration.fetch_toggles_uri.to_s,
+          headers: headers,
+          read_timeout: 60, # start a new SSE connection when no heartbeat received in 1 minute
+          reconnect_time: 2,
+          connect_timeout: 10,
+          logger: Unleash.logger
+        )
+
+        self.event_source.on_event do |event|
+          handle_event(event)
+        end
+
+        self.event_source.on_error do |error|
+          Unleash.logger.warn "Streaming error: #{error}"
+        end
+
+        self.running = true
+      end
+    end
+
+    def stop
+      self.mutex.synchronize do
+        return unless self.running
+
+        Unleash.logger.info "Stopping streaming client"
+        self.running = false
+        self.event_source&.close
+        self.event_source = nil
+      end
+    end
+
+    def running?
+      self.mutex.synchronize{ self.running }
+    end
+
+    private
+
+    def handle_event(event)
+      case event.type.to_s
+      when 'unleash-connected'
+        Unleash.logger.debug "Streaming client connected"
+        handle_connected_event(event)
+      when 'unleash-updated'
+        Unleash.logger.debug "Received streaming update"
+        handle_updated_event(event)
+      else
+        Unleash.logger.debug "Received unknown event type: #{event.type}"
+      end
+    rescue StandardError => e
+      Unleash.logger.error "Error handling streaming event: #{e.message}"
+    end
+
+    def handle_connected_event(event)
+      Unleash.logger.debug "Processing initial hydration data"
+      handle_updated_event(event)
+    end
+
+    def handle_updated_event(event)
+      self.mutex.synchronize do
+        self.toggle_engine.take_state(event.data)
+      end
+
+      # TODO: update backup file
+    rescue JSON::ParserError => e
+      Unleash.logger.error "Failed to parse streaming event data: #{e.message}"
+    rescue StandardError => e
+      Unleash.logger.error "Error processing delta update: #{e.message}"
+    end
+  end
+end

spec/unleash/configuration_spec.rb

@@ -105,6 +105,12 @@
       expect(config.fetch_toggles_uri.to_s).to eq('https://testurl/api/client/features?project=test-project')
     end
 
+    it "should build the correct unleash endpoint when streaming mode is enabled" do
+      config = Unleash::Configuration.new(url: 'https://testurl/api', app_name: 'test-app')
+      config.experimental_mode = { type: 'streaming' }
+      expect(config.fetch_toggles_uri.to_s).to eq('https://testurl/api/client/streaming')
+    end
+
     it "should allow hashes for custom_http_headers via yield" do
       Unleash.configure do |config|
         config.url      = 'http://test-url/'

unleash-client.gemspec

@@ -23,6 +23,7 @@ Gem::Specification.new do |spec|
   spec.require_paths = ["lib"]
   spec.required_ruby_version = ">= 2.7"
 
+  spec.add_dependency "ld-eventsource", "~> 2.2.0"
   spec.add_dependency "yggdrasil-engine", "~> 1.0.4"
 
   spec.add_dependency "base64", "~> 0.3.0"