Class: OmniAI::Google::Live::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/omniai/google/live/connection.rb

Overview

A WebSocket client over TCP or TLS, driven by websocket-driver with one reader thread per connection.

Callbacks run on the reader thread. Writes may come from any thread: they are serialized by a mutex and bounded by write_timeout, so a stalled peer can't block the caller indefinitely. An idle connection is pinged, so a silently dropped path is noticed. on_close is called exactly once, whether the peer closes, the socket hits EOF or errors, a write or pong times out, or a callback raises.

Defined Under Namespace

Classes: ConnectTimeout, PongTimeout, WriteTimeout

Constant Summary collapse

CONNECT_TIMEOUT =

seconds, TCP and TLS together

10
CLOSE_TIMEOUT =

seconds

2
PING_INTERVAL =

seconds without a frame before pinging

15
PONG_TIMEOUT =

seconds

10
READ_CHUNK =

bytes

16_384

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(url:, headers:, write_timeout:, on_open:, on_message:, on_close:) ⇒ Connection

Returns a new instance of Connection.

Parameters:

  • url (String) —

    a ws:// or wss:// URL

  • headers (Hash) —

    sent with the upgrade request

  • write_timeout (Numeric) —

    seconds

  • on_open (Proc) —

    called with no arguments

  • on_message (Proc) —

    called with the payload String

  • on_close (Proc) —

    called with (code, reason, error)



35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/omniai/google/live/connection.rb', line 35

def initialize(url:, headers:, write_timeout:, on_open:, on_message:, on_close:)
  @url = url
  @headers = headers
  @write_timeout = write_timeout
  @on_open = on_open
  @on_message = on_message
  @on_close = on_close
  @mutex = Mutex.new
  @pending = []
  @finished = false
end

Instance Attribute Details

#url ⇒ String (readonly)

Returns:

  • (String) 


27
28
29
 
# File 'lib/omniai/google/live/connection.rb', line 27

def url
  @url
end

Instance Method Details

#close ⇒ Object

Sends a close frame and waits briefly for the peer's reply before dropping the socket. Called from a callback (the reader thread itself), it only sends the frame: the reader finishes when the reply arrives.



82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/omniai/google/live/connection.rb', line 82

def close
  return if @finished || @reader.nil?

  begin
    @mutex.synchronize { @driver.close }
  rescue WriteTimeout, IOError, SystemCallError
    nil
  end
  return if Thread.current == @reader

  abort! unless @reader.join(CLOSE_TIMEOUT)
  @reader.join(CLOSE_TIMEOUT)
end

#connect ⇒ Object

Raises:



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/omniai/google/live/connection.rb', line 48

def connect
  @io = open_socket
  @wake_reader, @wake_writer = IO.pipe # for abort!: on Linux, closing a socket doesn't interrupt a select on it
  @driver = WebSocket::Driver.client(self, binary_data_format: :string)
  @headers.each { |name, value| @driver.set_header(name, value) }
  @driver.on(:open) { @pending << [:open] }
  @driver.on(:message) { |event| @pending << [:message, event.data] }
  @driver.on(:close) { |event| @pending << [:close, event.code, event.reason] }
  @mutex.synchronize { @driver.start }
  @reader = Thread.new { read_loop }
  @reader.name = "omniai-google-live"
  @reader.report_on_exception = false
rescue StandardError
  [@io, @wake_reader, @wake_writer].compact.each { |io| close_quietly(io) } unless @reader
  raise
end

#open? ⇒ Boolean

Returns:

  • (Boolean) 


66
67
68
 
# File 'lib/omniai/google/live/connection.rb', line 66

def open?
  !@finished && !@aborted
end

#send_text(payload) ⇒ Object

Parameters:

  • payload (String)

Raises:



73
74
75
76
77
78
 
# File 'lib/omniai/google/live/connection.rb', line 73

def send_text(payload)
  @mutex.synchronize { @driver.text(payload) }
rescue WriteTimeout => e
  abort!(e)
  raise
end

#write(data) ⇒ Object

Called by websocket-driver (under the mutex) for every outgoing frame.

Parameters:

  • data (String) 


99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/omniai/google/live/connection.rb', line 99

def write(data)
  deadline = monotonic + @write_timeout
  data = data.b
  until data.empty?
    result = @io.write_nonblock(data, exception: false)
    case result
    when :wait_writable then wait!(nil, [@io.to_io], deadline) { write_timeout }
    when :wait_readable then wait!([@io.to_io], nil, deadline) { write_timeout }
    else data = data.byteslice(result..)
    end
  end
end