Class: SplitIoClient::SSE::EventSource::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/splitclient-rb/sse/event_source/client.rb

Constant Summary collapse

DEFAULT_READ_TIMEOUT = 
70
CONNECT_TIMEOUT = 
30
OK_CODE = 
200
KEEP_ALIVE_RESPONSE = 
"c\r\n:keepalive\n\n\r\n".freeze
ERROR_EVENT_TYPE = 
'error'.freeze

Instance Method Summary collapse

Constructor Details

#initialize(config, api_key, telemetry_runtime_producer, event_parser, notification_manager_keeper, notification_processor, status_queue, read_timeout: DEFAULT_READ_TIMEOUT) ⇒ Client

Returns a new instance of Client.



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/splitclient-rb/sse/event_source/client.rb', line 18

def initialize(config,
               api_key,
               telemetry_runtime_producer,
               event_parser,
               notification_manager_keeper,
               notification_processor,
               status_queue,
               read_timeout: DEFAULT_READ_TIMEOUT)
  @config = config
  @api_key = api_key
  @telemetry_runtime_producer = telemetry_runtime_producer
  @event_parser = event_parser
  @notification_manager_keeper = notification_manager_keeper
  @notification_processor = notification_processor
  @status_queue = status_queue
  @read_timeout = read_timeout
  @connected = Concurrent::AtomicBoolean.new(false)
  # Set while close is tearing a connection down on purpose, so the reader
  # thread does not report the resulting socket error as a retryable failure.
  @shutdown = Concurrent::AtomicBoolean.new(false)
  # Incremented for every connection attempt. A reader thread whose generation
  # is no longer current has been superseded and must exit without side effects.
  @generation = Concurrent::AtomicFixnum.new(0)
  @socket = nil
  # Serialises concurrent close calls so the second caller does not
  # re-push the status or double-close a socket the first caller
  # already claimed.
  @close_mutex = Mutex.new
end

Instance Method Details

#close(status = nil) ⇒ Object 



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/splitclient-rb/sse/event_source/client.rb', line 48

def close(status = nil)
  socket = nil
  @close_mutex.synchronize do
    return if @socket.nil?

    @config.logger.debug('Closing SSEClient socket') if @config.debug_enabled
    @shutdown.make_true
    push_status(status)
    @connected.make_false

    socket = @socket
    @socket = nil
  end

  socket.sync_close = true if socket.is_a? OpenSSL::SSL::SSLSocket
  socket.close
  @config.logger.debug("SSEClient socket state #{socket.state}") if socket.is_a?(OpenSSL::SSL::SSLSocket) && @config.debug_enabled
rescue StandardError => e
  @config.logger.error("SSEClient close Error: #{e.inspect}")
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


90
91
92
 
# File 'lib/splitclient-rb/sse/event_source/client.rb', line 90

def connected?
  @connected.value
end

#start(url) ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/splitclient-rb/sse/event_source/client.rb', line 69

def start(url)
  if connected?
    @config.logger.debug('SSEClient already running.') if @config.debug_enabled
    return true
  end

  @uri = URI(url)
  latch = Concurrent::CountDownLatch.new(1)
  connect_thread(latch)

  unless latch.wait(CONNECT_TIMEOUT)
    @config.logger.warn("SSE connection attempt did not complete within #{CONNECT_TIMEOUT} seconds.")
    return false
  end

  connected?
rescue StandardError => e
  @config.logger.error("SSEClient start Error: #{e.inspect}")
  connected?
end