Class: ERPC::WebSocketJsonRpcTransport

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

Instance Method Summary collapse

Constructor Details

#initialize(connection_url, credential, timeout, connection_factory: nil) ⇒ WebSocketJsonRpcTransport

Returns a new instance of WebSocketJsonRpcTransport.



208
209
210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/erpc_sdk/websocket.rb', line 208

def initialize(connection_url, credential, timeout, connection_factory: nil)
  @connection_url = connection_url
  @credential = credential
  @timeout = timeout
  @connection_factory = connection_factory || ->(url, seconds) { WebSocketConnection.new(url, seconds) }
  @mutex = Mutex.new
  @connection = nil
  @reader = nil
  @pending = {}
  @listeners = {}
  @next_id = 0
  @next_listener_id = 0
  @closed = false
end

Instance Method Details

#close ⇒ Object 



253
254
255
256
257
258
259
260
261
262
263
264
265
266
 
# File 'lib/erpc_sdk/websocket.rb', line 253

def close
  connection = @mutex.synchronize do
    return if @closed

    @closed = true
    current, @connection = @connection, nil
    current
  end
  connection&.close
  @reader&.join(1)
  fail_pending(TransportError.new("WebSocket transport is closed"))
  @mutex.synchronize { @listeners.clear }
  nil
end

#on_notification(listener = nil, &block) ⇒ Object

Raises:

  • (ArgumentError) 


241
242
243
244
245
246
247
248
249
250
251
 
# File 'lib/erpc_sdk/websocket.rb', line 241

def on_notification(listener = nil, &block)
  selected = listener || block
  raise ArgumentError, "listener is required" unless selected

  listener_id = @mutex.synchronize do
    @next_listener_id += 1
    @listeners[@next_listener_id] = selected
    @next_listener_id
  end
  -> { @mutex.synchronize { @listeners.delete(listener_id) } }
end

#request(method, params = nil) ⇒ Object 



223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/erpc_sdk/websocket.rb', line 223

def request(method, params = nil)
  connection = ensure_connection
  request_id = @mutex.synchronize { @next_id += 1 }
  queue = Queue.new
  @mutex.synchronize { @pending[request_id] = queue }
  body = { "jsonrpc" => "2.0", "id" => request_id, "method" => method }
  body["params"] = params unless params.nil?
  connection.write_text(JSON.generate(body))
  response = Timeout.timeout(@timeout) { queue.pop }
  raise response if response.is_a?(Exception)

  unwrap(response, request_id)
rescue ::Timeout::Error
  raise TimeoutError, @timeout
ensure
  @mutex.synchronize { @pending.delete(request_id) } if request_id
end