Module: PumaPlus::WS

Defined in:
lib/puma_plus/ws.rb

Overview

Outbound WebSocket API.

The application never touches a socket. Go owns every WebSocket connection; this module sends it commands over a per-process control channel, which means the app can address a connection at any time from any thread -- not only while handling a message from it.

PumaPlus::WS.send(conn_id, "pong")
PumaPlus::WS.subscribe(conn_id, "room:42")
PumaPlus::WS.publish("room:42", payload)   # one call -> N sockets
PumaPlus::WS.close(conn_id, code: 1000)

#publish is the reason this exists. Go does the fan-out on goroutines, so a broadcast to ten thousand subscribers costs the application exactly one frame instead of ten thousand. Doing that in Ruby is what currently pushes ActionCable into a separate process with Redis behind it.

Defined Under Namespace

Classes: NotConnected

Constant Summary collapse

TEXT = 
"text"
BINARY = 
"binary"

Class Method Summary collapse

Class Method Details

.close(conn_id, code: 1000, reason: "") ⇒ Object 



123
124
125
126
 
# File 'lib/puma_plus/ws.rb', line 123

def close(conn_id, code: 1000, reason: "")
  command(Wire::WS_CLOSE,
          { "conn_id" => conn_id, "code" => code.to_s, "reason" => reason })
end

.connect!(socket_path, logger: $stderr) ⇒ Object

Called once per worker process, before any threads start.



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/puma_plus/ws.rb', line 32

def connect!(socket_path, logger: $stderr)
  @logger = logger
  @socket_path = socket_path
  @monitor = Monitor.new
  @pending = {}
  @seq = 0
  @conn = UNIXSocket.new(socket_path)
  @conn.sync = true
  @conn.write Wire.frame(Wire::HELLO, Wire.encode_kv([
    ["version", Wire::VERSION],
    ["role", "ws"],
    ["pid", Process.pid]
  ]))
  start_reader
  true
rescue SystemCallError => e
  @conn = nil
  logger.puts "[puma-plus] ws control channel unavailable: #{e.class}: #{e.message}"
  false
end

.connected? ⇒ Boolean

Is the control channel usable?

Returns:

  • (Boolean) 


54
 
# File 'lib/puma_plus/ws.rb', line 54

def connected? = !@conn.nil?

.connections ⇒ Object

Total live WebSocket connections across the whole server, not just this worker -- Go holds them all.



135
136
137
 
# File 'lib/puma_plus/ws.rb', line 135

def connections
  query("connections").to_i
end

.datagram(conn_id, payload) ⇒ Object

Send a best-effort datagram to one connection. WebTransport only.

Deliberately a separate method from #send rather than a keyword on it. A datagram may be dropped, duplicated or delivered out of order; that is a fundamentally different contract, and a one-word argument is far too easy to copy between call sites without noticing the guarantee changed.

Silently does nothing for a WebSocket client, which has no unreliable channel. Check PumaPlus::WSEvent#transport if you need to know.



76
77
78
 
# File 'lib/puma_plus/ws.rb', line 76

def datagram(conn_id, payload)
  command(Wire::WS_DATAGRAM, { "conn_id" => conn_id }, payload)
end

.open?(conn_id) ⇒ Boolean

Is this connection still open?

Returns:

  • (Boolean) 


140
141
142
 
# File 'lib/puma_plus/ws.rb', line 140

def open?(conn_id)
  query("exists", "conn_id" => conn_id) == "true"
end

.open_stream(conn_id) ⇒ Object

Open a server-initiated stream on a WebTransport session. Returns the new stream id, or nil if the transport has no streams.



82
83
84
85
 
# File 'lib/puma_plus/ws.rb', line 82

def open_stream(conn_id)
  id = query_raw(Wire::WS_OPEN_STREAM, "conn_id" => conn_id)
  id.nil? || id.empty? ? nil : id
end

.publish(topic, payload, opcode: TEXT) ⇒ Object

Broadcast to every subscriber of a topic.

One frame regardless of how many subscribers there are; Go performs the writes. Returns nil rather than a delivery count because waiting for one would serialise the caller behind every recipient's socket -- use #subscribers if you need the number.



93
94
95
 
# File 'lib/puma_plus/ws.rb', line 93

def publish(topic, payload, opcode: TEXT)
  command(Wire::WS_PUBLISH, { "topic" => topic, "opcode" => opcode }, payload)
end

.publish_datagram(topic, payload) ⇒ Object

Broadcast a best-effort datagram to a topic. WebTransport subscribers receive it; WebSocket subscribers are skipped, since they have no unreliable channel.

WS.publish("room:42", chat)            # reliable -- everyone
WS.publish_datagram("game:7", pos)     # lossy -- WebTransport only

The asymmetry is the point: one of those is safe for chat, the other is not.



106
107
108
 
# File 'lib/puma_plus/ws.rb', line 106

def publish_datagram(topic, payload)
  command(Wire::WS_PUBLISH_DATAGRAM, { "topic" => topic }, payload)
end

.send(conn_id, payload, opcode: TEXT, stream: nil) ⇒ Object

Send a reliable, ordered message to one connection.

stream: targets one stream of a WebTransport session. Without it the message goes to the connection's default channel, which is all a WebSocket has anyway -- so code that ignores streams works on both.



61
62
63
64
65
 
# File 'lib/puma_plus/ws.rb', line 61

def send(conn_id, payload, opcode: TEXT, stream: nil)
  args = { "conn_id" => conn_id, "opcode" => opcode }
  args["stream_id"] = stream if stream
  command(Wire::WS_SEND, args, payload)
end

.subscribe(conn_id, topic) ⇒ Object 



115
116
117
 
# File 'lib/puma_plus/ws.rb', line 115

def subscribe(conn_id, topic)
  command(Wire::WS_SUBSCRIBE, { "conn_id" => conn_id, "topic" => topic })
end

.subscribers(topic) ⇒ Object

How many connections are subscribed to a topic. Synchronous.



129
130
131
 
# File 'lib/puma_plus/ws.rb', line 129

def subscribers(topic)
  query("subscribers", "topic" => topic).to_i
end

.transport(conn_id) ⇒ Object

Which transport a connection is using: "websocket" or "webtransport".



111
112
113
 
# File 'lib/puma_plus/ws.rb', line 111

def transport(conn_id)
  query("transport", "conn_id" => conn_id)
end

.unsubscribe(conn_id, topic) ⇒ Object 



119
120
121
 
# File 'lib/puma_plus/ws.rb', line 119

def unsubscribe(conn_id, topic)
  command(Wire::WS_UNSUBSCRIBE, { "conn_id" => conn_id, "topic" => topic })
end