Class: SchwabRb::Stream::Base

Inherits:
Object
  • Object
show all
Defined in:
lib/schwab_rb/stream/base.rb

Constant Summary collapse

INITIAL_BACKOFF = 
2
MAX_BACKOFF = 
120
MIN_CONNECTION_TIME = 
90

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client) ⇒ Base

Returns a new instance of Base.



19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/schwab_rb/stream/base.rb', line 19

def initialize(client)
  @client = client
  @handlers = {}
  @subscriptions = {}
  @connection = nil
  @message_builder = nil
  @streamer_info = nil
  @connected = false
  @connect_time = nil
  @backoff_time = INITIAL_BACKOFF
  @should_run = false
end

Instance Attribute Details

#client ⇒ Object (readonly)

Returns the value of attribute client.



17
18
19
 
# File 'lib/schwab_rb/stream/base.rb', line 17

def client
  @client
end

Instance Method Details

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


70
71
72
 
# File 'lib/schwab_rb/stream/base.rb', line 70

def connected?
  @connected
end

#on(service_symbol, symbols: nil, fields: nil, &block) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/schwab_rb/stream/base.rb', line 32

def on(service_symbol, symbols: nil, fields: nil, &block)
  service_name = Services.lookup(service_symbol)
  resolved_fields = fields ? Fields.resolve(service_name, fields) : nil

  handler = { symbols: Array(symbols), fields: resolved_fields, callback: block }
  @handlers[service_name] ||= []
  @handlers[service_name] << handler

  @subscriptions[service_name] = merge_subscription(
    @subscriptions[service_name],
    handler
  )

  send_subscription(service_name, handler) if @connected

  self
end

#start ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/schwab_rb/stream/base.rb', line 50

def start
  @should_run = true
  Async do |task|
    loop do
      connect
      run_receive_loop
    rescue StandardError => e
      handle_disconnect(e)
      break unless should_reconnect?

      wait_and_reconnect(task)
    end
  end
end

#stop ⇒ Object 



65
66
67
68
 
# File 'lib/schwab_rb/stream/base.rb', line 65

def stop
  @should_run = false
  disconnect
end