Class: LaunchDarkly::Impl::DataSource::StreamProcessor Private

Inherits:
Object
  • Object
show all
Defined in:
lib/ldclient-rb/impl/data_source/stream.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Since:

  • 5.5.0

Instance Method Summary collapse

Constructor Details

#initialize(sdk_key, config, diagnostic_accumulator = nil) ⇒ StreamProcessor

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of StreamProcessor.

Since:

  • 5.5.0



25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/ldclient-rb/impl/data_source/stream.rb', line 25

def initialize(sdk_key, config, diagnostic_accumulator = nil)
  @sdk_key = sdk_key
  @config = config
  @diagnostic_accumulator = diagnostic_accumulator
  @data_source_update_sink = config.data_source_update_sink
  @feature_store = config.feature_store
  @initialized = Concurrent::AtomicBoolean.new(false)
  @started = Concurrent::AtomicBoolean.new(false)
  @stopped = Concurrent::AtomicBoolean.new(false)
  @ready = Concurrent::Event.new
  @stop_event = Concurrent::Event.new
  @retry_state = Impl::RetryState.for_streaming(@config.initial_reconnect_delay, @config.logger)
  @connection_attempt_start_time = 0
  @connection_attempt_started_monotonic = nil
end

Instance Method Details

#initialized? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Boolean)

Since:

  • 5.5.0



41
42
43
 
# File 'lib/ldclient-rb/impl/data_source/stream.rb', line 41

def initialized?
  @initialized.value
end

#start ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 5.5.0



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/ldclient-rb/impl/data_source/stream.rb', line 45

def start
  return @ready unless @started.make_true

  @config.logger.info { "[LDClient] Initializing stream connection" }

  headers = Impl::Util.default_http_headers(@sdk_key, @config)
  opts = {
    headers: headers,
    read_timeout: READ_TIMEOUT_SECONDS,
    connect_timeout: @config.connect_timeout,
    logger: @config.logger,
    socket_factory: @config.socket_factory,
    # The SDK waits in its error handler before each reconnect, so the SSE client adds no delay.
    reconnect_time: 0,
  }
  log_connection_started

  uri = Impl::Util.add_payload_filter_key(@config.stream_uri + "/all", @config)
  @es = SSE::Client.new(uri, **opts) do |conn|
    conn.on_connect { |response_headers| DataSource.record_environment_id(@data_source_update_sink, response_headers) }
    conn.on_event { |event| process_message(event) }
    conn.on_error { |err| handle_error(err) }
  end
  # stop can run before @es is set, and then it has no client to close.
  @es.close if @stopped.value

  @ready
end

#stop ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 5.5.0



74
75
76
 
# File 'lib/ldclient-rb/impl/data_source/stream.rb', line 74

def stop
  stop_with_error_info
end