Module: Cybertrain::WebSocket

Defined in:
cybertrain/websocket.rb,
cybertrain/websocket/frame.rb,
cybertrain/websocket/client.rb

Defined Under Namespace

Modules: Frame Classes: BlockedAddressError, Client, ClosedError, Connection, Error, HandshakeError, Message, ProtocolError, RefusedError, Status, Subscription, TLSError, TimeoutError, TooLargeError

Class Method Summary collapse

Class Method Details

.feed_status(name) ⇒ Status

The state of the feed named name. Not part of /up: an application’s own health check can use it.

Parameters:

  • name (String)

Returns:

Raises:

  • (ArgumentError) —

    when no feed has that name



335
336
337
338
339
340
341
 
# File 'cybertrain/websocket.rb', line 335

def self.feed_status(name)
  i = subscription_index(name.to_s)
  raise ArgumentError, "WebSocket.feed_status: no subscription named #{name.to_s.inspect}" if i < 0

  sub = SUBSCRIPTIONS[i]
  Status.new(sub.sub_connected?, sub.sub_last_message_at, sub.sub_reconnects)
end

.random_source=(source) ⇒ Proc

The source of the retry delay’s jitter: a Proc answering a Float in 0.0..1.0. Tests set -> { 0.5 }.

Parameters:

  • source (Proc)

Returns:

  • (Proc) 


451
452
453
 
# File 'cybertrain/websocket.rb', line 451

def self.random_source=(source)
  @random_source = source
end

.retry_delay(attempt, jitter) ⇒ Float

Seconds to wait before connecting again after the attemptth failure in a row: min(60, 2 ** (attempt - 1)), times 0.8 to 1.2.

Parameters:

  • attempt (Integer) —

    1 for the first failure

  • jitter (Float) —

    0.0..1.0 (0.5 is the plain doubling)

Returns:

  • (Float) 


431
432
433
434
435
436
437
438
439
440
441
442
443
444
 
# File 'cybertrain/websocket.rb', line 431

def self.retry_delay(attempt, jitter)
  n = attempt.to_i
  base = 1
  if n >= 7
    base = 60
  elsif n > 1
    base = 2**(n - 1)
  end
  j = jitter.to_f
  j = 0.0 if j < 0.0
  j = 1.0 if j > 1.0
  # Tenths, so (1, 1.0) is exactly 1.2 (0.8 + 0.4 is not).
  (base.to_f * (8.0 + 4.0 * j)) / 10.0
end

.running? ⇒ Boolean

Returns true between start! and stop!.

Returns:



421
422
423
 
# File 'cybertrain/websocket.rb', line 421

def self.running?
  @running
end

.start!(config = Cybertrain.config) ⇒ Boolean

Starts one thread per subscription, unless Config#worker is off, nothing is subscribed or the feeds already run. Application#serve calls it (through the Background service).

Parameters:

  • config (Config) (defaults to: Cybertrain.config)

Returns:

  • (Boolean) —

    true when it started the threads

Raises:

  • (ArgumentError) —

    when a subscription has no on_message



366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
 
# File 'cybertrain/websocket.rb', line 366

def self.start!(config = Cybertrain.config)
  return false if @running
  unless config.worker
    Cybertrain.logger.info("Background worker off (CYBERTRAIN_WORKER=0): WebSocket subscriptions run elsewhere")
    return false
  end
  return false if SUBSCRIPTIONS.empty?

  problem = subscriptions_problem
  raise ArgumentError, "WebSocket.start!: #{problem}" unless problem.empty?

  # spikes/NOTES.md rule 22: before the first Thread.new.
  ENV["SPINEL_WORKERS"] = "1" unless ENV["SPINEL_WORKERS"]
  @running = true
  @generation += 1
  generation = @generation
  i = 0
  while i < SUBSCRIPTIONS.size
    sub = SUBSCRIPTIONS[i]
    Cybertrain.logger.info("WebSocket started: #{sub.sub_name} (#{sub.sub_origin})")
    sub.sub_thread!(spawn_feed(sub, generation))
    i += 1
  end
  true
end

.stop! ⇒ nil

Asks every feed to end and waits for them, at most 2 s in all: a connected feed sends close 1000 and waits at most 1 s for the server’s answer. Every feed reads as disconnected once it returns (a thread still closing finishes on its own). Does nothing when the feeds do not run.

Returns:

  • (nil) 


399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
 
# File 'cybertrain/websocket.rb', line 399

def self.stop!
  return nil unless @running

  @running = false
  ends_at = Net::Timer.now + STOP_JOIN_SECONDS
  i = 0
  while i < SUBSCRIPTIONS.size
    thread = SUBSCRIPTIONS[i].sub_thread
    left = ends_at - Net::Timer.now
    thread.join(left) if !thread.nil? && left > 0.0
    i += 1
  end
  i = 0
  while i < SUBSCRIPTIONS.size
    SUBSCRIPTIONS[i].sub_disconnected!
    i += 1
  end
  nil
end

.subscribe(name, url, headers: Client::NO_HEADERS, timeout: 5, max_message: 1_048_576, ping_interval: 15, stale_after: 45, allow_private: Cybertrain.config.http_allow_private) ⇒ Subscription

Registers a feed to keep connected while the server runs. Register before the server starts; give it Cybertrain::WebSocket::Subscription#on_message (and usually Cybertrain::WebSocket::Subscription#on_open).

Examples:

Cybertrain::WebSocket.subscribe("trades", "wss://feed.example.com/v1")
  .on_open { |conn| conn.send_text("{\"op\":\"subscribe\"}") }
  .on_message { |conn, text| Trade.record(text) }

Parameters:

  • name (String) —

    for the log and feed_status; unique

  • url (String) —

    ws:// or wss://

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

    extra handshake headers

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

    seconds per wait while connecting and per write

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

    the largest message accepted, in bytes

  • ping_interval (Integer) (defaults to: 15) —

    seconds of quiet before a ping, at least 1

  • stale_after (Integer) (defaults to: 45) —

    seconds of quiet before reconnecting, more than ping_interval

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

    allow non-public destinations

Returns:

Raises:

  • (ArgumentError) —

    an empty or taken name, a URL that is not ws:// or wss://, a header the client refuses, a bad interval



296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
 
# File 'cybertrain/websocket.rb', line 296

def self.subscribe(name, url, headers: Client::NO_HEADERS, timeout: 5, max_message: 1_048_576,
                   ping_interval: 15, stale_after: 45,
                   allow_private: Cybertrain.config.http_allow_private)
  label = name.to_s
  raise ArgumentError, "WebSocket.subscribe: the name is empty" if label.empty?
  raise ArgumentError, "WebSocket.subscribe #{label.inspect}: already registered" if subscription_index(label) >= 0

  target = HttpClient.parse_url_for(url, "ws://", "wss://")
  ping = ping_interval.to_i
  stale = stale_after.to_i
  raise ArgumentError, "WebSocket.subscribe #{label.inspect}: ping_interval must be at least 1" if ping < 1
  raise ArgumentError, "WebSocket.subscribe #{label.inspect}: stale_after must be more than ping_interval" if stale <= ping
  raise ArgumentError, "WebSocket.subscribe #{label.inspect}: timeout must be positive" unless timeout.to_f > 0.0
  raise ArgumentError, "WebSocket.subscribe #{label.inspect}: max_message must be at least 1" if max_message.to_i < 1

  # The headers are checked now, not at every connect.
  Client.handshake_head(target, "", headers)
  sub = Subscription.new(label, url.to_s, target, headers, timeout.to_f, max_message.to_i, ping, stale, allow_private == true)
  SUBSCRIPTIONS << sub
  sub
end

.subscriptions_problem ⇒ String

Why start! would refuse to start: the first subscription without on_message, named. Application#serve asks it (through Background.boot_problem) before binding the port.

Returns:

  • (String) —

    ”” when every subscription has on_message



348
349
350
351
352
353
354
355
356
357
 
# File 'cybertrain/websocket.rb', line 348

def self.subscriptions_problem
  i = 0
  while i < SUBSCRIPTIONS.size
    sub = SUBSCRIPTIONS[i]
    return "WebSocket subscription #{sub.sub_name.inspect} has no on_message" if sub.sub_message_block.nil?

    i += 1
  end
  ""
end