Class: DhanHQ::WS::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/DhanHQ/ws/client.rb

Overview

Client responsible for managing the lifecycle of a streaming connection to the DhanHQ market data WebSocket.

The client encapsulates reconnection logic, subscription state tracking, and event dispatching. It is typically used indirectly via connect, but can also be instantiated directly for more advanced flows.

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(mode: :ticker, url: nil) ⇒ Client

Returns a new instance of Client.

Parameters:

  • mode (Symbol) (defaults to: :ticker) —

    Feed mode (:ticker, :quote, :full).

  • url (String, nil) (defaults to: nil) —

    Optional custom WebSocket endpoint.

Raises:



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/DhanHQ/ws/client.rb', line 23

def initialize(mode: :ticker, url: nil)
  @mode  = mode # :ticker, :quote, :full (adjust to your API)
  @bus   = CmdBus.new
  @state = SubState.new
  @callbacks = Concurrent::Map.new { |h, k| h[k] = [] }
  @started = Concurrent::AtomicBoolean.new(false)

  token = DhanHQ.configuration.resolved_access_token
  raise DhanHQ::AuthenticationError, "Missing access token" if token.nil? || token.empty?

  cid   = DhanHQ.configuration.client_id or raise "DhanHQ.client_id not set"
  ver   = (DhanHQ.configuration.respond_to?(:ws_version) && DhanHQ.configuration.ws_version) || 2
  base  = url || DhanHQ.configuration.ws_market_feed_url
  @url  = base.include?("?") ? base : "#{base}?version=#{ver}&token=#{token}&clientId=#{cid}&authType=2"
end

Class Method Details

.install_at_exit_hook! ⇒ void

This method returns an undefined value.

Installs a single at_exit hook to close open WebSocket clients.



143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/DhanHQ/ws/client.rb', line 143

def self.install_at_exit_hook!
  return if defined?(@_at_exit_installed) && @_at_exit_installed

  @_at_exit_installed = true
  at_exit do
    DhanHQ.logger&.info("[DhanHQ::WS] at_exit: disconnecting all local clients")
    Registry.stop_all
  rescue StandardError => e
    DhanHQ.logger&.debug("[DhanHQ::WS] at_exit error #{e.class}: #{e.message}")
  end
end

Instance Method Details

#connected? ⇒ Boolean

Indicates whether the underlying WebSocket connection is open.

Returns:

  • (Boolean) 


87
88
89
90
91
 
# File 'lib/DhanHQ/ws/client.rb', line 87

def connected?
  return false unless @started.true?

  @conn&.open? || false
end

#disconnect! ⇒ DhanHQ::WS::Client

Immediately disconnects from the feed by sending the disconnect frame (RequestCode 12) before closing the socket.

Returns:



74
75
76
77
78
79
80
81
82
 
# File 'lib/DhanHQ/ws/client.rb', line 74

def disconnect!
  return self unless @started.true?

  @started.make_false
  @conn&.disconnect!
  Registry.unregister(self)
  emit(:close, true)
  self
end

#on(event) {|payload| ... } ⇒ DhanHQ::WS::Client

Registers a callback for a given event.

Parameters:

  • event (Symbol) —

    Event name (:tick, :open, :close, :error).

Yield Parameters:

  • payload (Object) —

    Event payload.

Returns:



160
161
162
163
 
# File 'lib/DhanHQ/ws/client.rb', line 160

def on(event, &blk)
  @callbacks[event] << blk
  self
end

#start ⇒ DhanHQ::WS::Client

Starts the WebSocket connection and event loop.

Returns:



42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/DhanHQ/ws/client.rb', line 42

def start
  return self if @started.true?

  @started.make_true
  @conn = Connection.new(url: @url, mode: @mode, bus: @bus, state: @state) do |binary|
    tick = Decoder.decode(binary)
    emit(:tick, tick) if tick
  end
  Registry.register(self)
  install_at_exit_once!
  @conn.start
  self
end

#stop ⇒ DhanHQ::WS::Client

Gracefully stops the connection without sending a manual disconnect frame.

Returns:



60
61
62
63
64
65
66
67
68
 
# File 'lib/DhanHQ/ws/client.rb', line 60

def stop
  return unless @started.true?

  @started.make_false
  @conn&.stop
  Registry.unregister(self)
  emit(:close, true)
  self
end

#subscribe_many(list) ⇒ DhanHQ::WS::Client

Subscribes to updates for a list of instruments.

Parameters:

  • list (Array<Hash>) —

    Array containing instrument hashes with :ExchangeSegment and :SecurityId keys.

Returns:



110
111
112
113
114
115
 
# File 'lib/DhanHQ/ws/client.rb', line 110

def subscribe_many(list)
  norms = Segments.normalize_instruments(list).map { |i| prune(i) }
  DhanHQ.logger&.info("[DhanHQ::WS] subscribe_many (normalized) -> #{norms}")
  @bus.sub(norms)
  self
end

#subscribe_one(segment:, security_id:) ⇒ DhanHQ::WS::Client

Subscribes to updates for a single instrument.

Parameters:

  • segment (String, Symbol, Integer)
  • security_id (String, Integer)

Returns:



98
99
100
101
102
103
 
# File 'lib/DhanHQ/ws/client.rb', line 98

def subscribe_one(segment:, security_id:)
  norm = Segments.normalize_instrument(ExchangeSegment: segment, SecurityId: security_id)
  DhanHQ.logger&.info("[DhanHQ::WS] subscribe_one (normalized) -> #{norm}")
  @bus.sub([prune(norm)])
  self
end

#unsubscribe_many(list) ⇒ DhanHQ::WS::Client

Removes the subscriptions for a list of instruments.

Parameters:

  • list (Array<Hash>) —

    Instrument definitions to unsubscribe.

Returns:



133
134
135
136
137
138
 
# File 'lib/DhanHQ/ws/client.rb', line 133

def unsubscribe_many(list)
  norms = Segments.normalize_instruments(list).map { |i| prune(i) }
  DhanHQ.logger&.info("[DhanHQ::WS] unsubscribe_many (normalized) -> #{norms}")
  @bus.unsub(norms)
  self
end

#unsubscribe_one(segment:, security_id:) ⇒ DhanHQ::WS::Client

Removes the subscription for a single instrument.

Parameters:

  • segment (String, Symbol, Integer)
  • security_id (String, Integer)

Returns:



122
123
124
125
126
127
 
# File 'lib/DhanHQ/ws/client.rb', line 122

def unsubscribe_one(segment:, security_id:)
  norm = Segments.normalize_instrument(ExchangeSegment: segment, SecurityId: security_id)
  DhanHQ.logger&.info("[DhanHQ::WS] unsubscribe_one (normalized) -> #{norm}")
  @bus.unsub([prune(norm)])
  self
end