Class: Cybertrain::WebSocket::Client

Inherits:
Object
  • Object
show all
Defined in:
cybertrain/websocket/client.rb

Overview

A WebSocket connection to a server, used from one thread.

Examples:

require "cybertrain/websocket/client"
require "cybertrain/net/tls"

ws = Cybertrain::WebSocket::Client.connect("wss://feed.example.com/v1/trades")
ws.send_text(JSON.generate({ "subscribe" => "BTC-USD" }))
while (msg = ws.receive(10))
  handle(msg.ws_data) if msg.ws_opcode == :text
end
ws.close

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.connect(url, headers: NO_HEADERS, timeout: 5, max_message: 1_048_576, allow_private: Cybertrain.config.http_allow_private) ⇒ Client

Connects and performs the opening handshake.

Parameters:

  • url (String) —

    ws:// or wss://

  • headers (Hash{String => String}) (defaults to: NO_HEADERS) —

    extra handshake headers (not Host, Upgrade, Connection, Sec-WebSocket-*, Content-Length, …)

  • timeout (Numeric) (defaults to: 5) —

    seconds for each wait while connecting and for each write; the handshake as a whole gets three times that

  • max_message (Integer) (defaults to: 1_048_576) —

    the largest message accepted, in bytes

  • allow_private (Boolean) (defaults to: Cybertrain.config.http_allow_private) —

    allow loopback, private and other non-public destinations. Default Config#http_allow_private.

Returns:

Raises:

  • (Error) —

    a subclass; ArgumentError for a URL or header the client refuses



63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'cybertrain/websocket/client.rb', line 63

def self.connect(url, headers: NO_HEADERS, timeout: 5, max_message: 1_048_576,
                 allow_private: Cybertrain.config.http_allow_private)
  target = HttpClient.parse_url_for(url, "ws://", "wss://")
  seconds = timeout.to_f
  limit = max_message.to_i
  raise ArgumentError, "timeout must be positive" unless seconds > 0.0
  raise ArgumentError, "max_message must be at least 1" if limit < 1

  key = Base64.strict_encode64(SecureRandom.random_bytes(16))
  head = Client.handshake_head(target, key, headers)
  Client.ws_open(target, key, head, seconds, limit, allow_private == true)
end

Instance Method Details

#close ⇒ nil

Sends close 1000, waits at most timeout for the server’s close and closes the connection. Safe to call twice; never raises.

Returns:

  • (nil) 


258
259
260
 
# File 'cybertrain/websocket/client.rb', line 258

def close
  ws_close_within(@timeout)
end

#last_activity_at ⇒ Float

When a frame last arrived (or the connection opened), on the monotonic clock (Net::Timer.now).

Returns:

  • (Float) 


196
197
198
 
# File 'cybertrain/websocket/client.rb', line 196

def last_activity_at
  @last_activity_at
end

#receive(wait) ⇒ Message?

The next data message, waiting at most wait seconds. Pings are answered and pongs noted on the way; once wait has passed it answers nil after the next control frame or fragment too, so a server that streams pings or pongs cannot keep it looping.

Parameters:

  • wait (Numeric) —

    seconds

Returns:

  • (Message, nil) —

    nil when the wait ends first

Raises:

  • (ClosedError) —

    the server closed (its code) or the connection ended (1006)

  • (ProtocolError) —

    the server broke the framing (close 1002 sent)

  • (TooLargeError) —

    a message past max_message (close 1009 sent)



246
247
248
249
250
251
252
 
# File 'cybertrain/websocket/client.rb', line 246

def receive(wait)
  ws_require_open
  seconds = wait.to_f
  @receive_ends_at = Net::Timer.now + seconds
  @reader.reader_timer = Net::Timer.new(@label, seconds, seconds)
  ws_receive_guarded
end

#send_binary(data) ⇒ nil

Sends a binary message in one frame.

Parameters:

  • data (String)

Returns:

  • (nil) 


220
221
222
 
# File 'cybertrain/websocket/client.rb', line 220

def send_binary(data)
  ws_send(Frame::OP_BINARY, data.to_s)
end

#send_ping(payload = "") ⇒ nil

Sends a ping; the server’s pong counts as activity.

Parameters:

  • payload (String) (defaults to: "") —

    at most 125 bytes

Returns:

  • (nil)

Raises:

  • (ArgumentError) 


228
229
230
231
232
233
 
# File 'cybertrain/websocket/client.rb', line 228

def send_ping(payload = "")
  data = payload.to_s
  raise ArgumentError, "a ping payload is at most 125 bytes" if data.bytesize > 125

  ws_send(Frame::OP_PING, data)
end

#send_text(text) ⇒ nil

Sends a text message in one frame.

Parameters:

  • text (String) —

    UTF-8

Returns:

  • (nil)

Raises:

  • (Error) —

    TimeoutError, ClosedError, …



212
213
214
 
# File 'cybertrain/websocket/client.rb', line 212

def send_text(text)
  ws_send(Frame::OP_TEXT, text.to_s)
end

#ws_label ⇒ String

“host:port”, what error messages name.

Returns:

  • (String) 


203
204
205
 
# File 'cybertrain/websocket/client.rb', line 203

def ws_label
  @label
end

#ws_open? ⇒ Boolean

Returns false once the connection is closed (by #close, by the server or by an error).

Returns:

  • (Boolean) —

    false once the connection is closed (by #close, by the server or by an error)



188
189
190
 
# File 'cybertrain/websocket/client.rb', line 188

def ws_open?
  @open
end