Class: Finlight::BaseWebSocketClient

Inherits:
Object
  • Object
show all
Defined in:
lib/finlight/websocket/base_client.rb

Overview

Shared implementation of the finlight streaming protocol: reconnect loop with exponential backoff, application-level ping/pong with watchdog, proactive connection rotation before the server-side lifetime cap, and optional duplicate suppression.

Use the concrete ArticleWebSocketClient and RawArticleWebSocketClient obtained from Client#websocket and Client#raw_websocket.

Defined Under Namespace

Classes: DriverSocket

Constant Summary collapse

RECENT_ARTICLE_CACHE_SIZE = 
10
CLOSE_GRACE_SECONDS = 
5
DIAL_RATE_LIMIT_BACKOFF_SECONDS = 
60
ERROR_RATE_LIMIT_BACKOFF_SECONDS = 
60
ERROR_BLOCKED_BACKOFF_SECONDS = 
3600
DEFAULT_ADMIN_KICK_RETRY_MS = 
900_000
CLOSE_BLOCKED =

Close codes of the finlight WebSocket protocol.

1008
CLOSE_PROACTIVE_ROTATION = 
4000
CLOSE_RATE_LIMITED = 
4001
CLOSE_USER_BLOCKED = 
4002
CLOSE_ADMIN_KICK = 
4003

Instance Method Summary collapse

Constructor Details

#initialize(config, ping_interval: 25, pong_timeout: 60, base_reconnect_delay: 0.5, max_reconnect_delay: 10.0, connection_lifetime: 115 * 60, takeover: false, on_close: nil, logger: Logging.default) ⇒ BaseWebSocketClient

Returns a new instance of BaseWebSocketClient.

Parameters:

  • config (Config)
  • ping_interval (Numeric) (defaults to: 25) —

    seconds between application-level pings

  • pong_timeout (Numeric) (defaults to: 60) —

    seconds without a pong before forcing a reconnect

  • base_reconnect_delay (Numeric) (defaults to: 0.5) —

    initial reconnect backoff in seconds

  • max_reconnect_delay (Numeric) (defaults to: 10.0) —

    backoff cap in seconds

  • connection_lifetime (Numeric) (defaults to: 115 * 60) —

    seconds before proactively rotating the connection (server caps connections at 2h)

  • takeover (Boolean) (defaults to: false) —

    take over an existing connection for the same key

  • on_close (#call, nil) (defaults to: nil) —

    callback invoked with (code, reason) whenever a connection closes



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/finlight/websocket/base_client.rb', line 59

def initialize(config, ping_interval: 25, pong_timeout: 60,
               base_reconnect_delay: 0.5, max_reconnect_delay: 10.0,
               connection_lifetime: 115 * 60, takeover: false, on_close: nil,
               logger: Logging.default)
  @config = config
  @ping_interval = ping_interval
  @pong_timeout = pong_timeout
  @base_reconnect_delay = base_reconnect_delay
  @max_reconnect_delay = max_reconnect_delay
  @connection_lifetime = connection_lifetime
  @takeover = takeover
  @on_close = on_close
  @logger = logger
  @stop = false
  @running = false
  @recent_articles = []
end

Instance Method Details

#connect(**params) {|article| ... } ⇒ Object

Connects to the stream and blocks, yielding every received article. Reconnects automatically (exponential backoff, honors server-mandated wait times) until #stop is called or the server preempts the connection in favor of a newer one.

Parameters:

  • params (Hash) —

    stream filters as snake_case keywords, e.g. query:, sources:, exclude_sources:, language:, tickers:, extended:, include_entities:, exclude_empty_content:, countries:, categories:, include_updates:

Yields:

  • (article) —

    every article received on the stream

Raises:

  • (BlockedError) —

    if the server permanently rejected the connection (close code 1008)



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/finlight/websocket/base_client.rb', line 89

def connect(**params, &on_article)
  raise ArgumentError, "finlight: connect requires a block" unless on_article
  raise Error, "finlight: connect is already running on this client" if @running

  @running = true
  @stop = false
  @reconnect_at = nil
  payload = Params.normalize(params)
  delay = @base_reconnect_delay

  begin
    until @stop
      @logger.info { "finlight: connecting to #{websocket_url}" }
      result = run_connection(payload, on_article)
      raise BlockedError if result == :blocked
      break if @stop

      delay = @base_reconnect_delay if result == :connected

      now = monotonic
      if @reconnect_at && @reconnect_at > now
        wait = @reconnect_at - now
        @logger.info { "finlight: waiting #{wait.round(1)}s until server-mandated reconnect time" }
      else
        wait = delay
        @logger.info { "finlight: reconnecting in #{wait.round(1)}s" }
        delay = [delay * 2, @max_reconnect_delay].min
      end
      interruptible_sleep(wait)
    end
  ensure
    @running = false
  end
  nil
end

#connect_async(**params, &on_article) ⇒ Thread

Runs #connect on a background thread and returns it. Joining the thread re-raises Finlight::BlockedError if the stream ended that way.

Returns:

  • (Thread) 


129
130
131
132
133
134
 
# File 'lib/finlight/websocket/base_client.rb', line 129

def connect_async(**params, &on_article)
  Thread.new do
    Thread.current.report_on_exception = false
    connect(**params, &on_article)
  end
end

#stop ⇒ Object

Stops the stream: the current connection closes and the reconnect loop ends. Safe to call from any thread; #connect returns shortly after.



138
139
140
141
 
# File 'lib/finlight/websocket/base_client.rb', line 138

def stop
  @stop = true
  nil
end