Class: OMQ::Transport::WebSocket::AcceptedConnection

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

Overview

Server-side wrapper around an Async::WebSocket::Connection. Delegates the WS interface used by Connection (+read+, send_binary, flush, protocol) and adds wait_for_close so the Adapters::HTTP.open fiber can block until the engine closes the connection.

close only resolves the promise — the actual WebSocket close is performed by Adapters::HTTP.open's ensure block once wait_for_close returns. Calling @ws.close here too would double-close and (on Ruby 3.3) deadlock when the peer has already closed.

Instance Method Summary collapse

Constructor Details

#initialize(ws) ⇒ AcceptedConnection

Returns a new instance of AcceptedConnection.



236
237
238
239
 
# File 'lib/omq/transport/websocket/transport.rb', line 236

def initialize(ws)
  @ws             = ws
  @closed_promise = ::Async::Promise.new
end

Instance Method Details

#close ⇒ Object 



262
263
264
 
# File 'lib/omq/transport/websocket/transport.rb', line 262

def close
  @closed_promise.resolve(true)
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


267
268
269
 
# File 'lib/omq/transport/websocket/transport.rb', line 267

def closed?
  @closed_promise.resolved?
end

#flush ⇒ Object 



257
258
259
 
# File 'lib/omq/transport/websocket/transport.rb', line 257

def flush
  @ws.flush
end

#protocol ⇒ Object 



242
243
244
 
# File 'lib/omq/transport/websocket/transport.rb', line 242

def protocol
  @ws.protocol
end

#read ⇒ Object 



247
248
249
 
# File 'lib/omq/transport/websocket/transport.rb', line 247

def read
  @ws.read
end

#send_binary(buffer, **opts) ⇒ Object 



252
253
254
 
# File 'lib/omq/transport/websocket/transport.rb', line 252

def send_binary(buffer, **opts)
  @ws.send_binary(buffer, **opts)
end

#wait_for_close ⇒ Object 



272
273
274
 
# File 'lib/omq/transport/websocket/transport.rb', line 272

def wait_for_close
  @closed_promise.wait
end