Class: OMQ::Transport::WebSocket::Connection

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

Overview

Quacks like Protocol::ZMTP::Connection but speaks ZWS 2.0 over an Async::WebSocket::Connection (or any object that responds to #read, #send_binary, #flush, #close, #protocol).

The HTTP/WS upgrade is already complete by the time this is constructed — the Engine instantiates this class via OMQ::Transport::WebSocket.connection_class once handle_accepted / handle_connected delivers the ws_conn.

No ZMTP/3.1 greeting. The mechanism (NULL or no-mechanism) is determined from the negotiated Sec-WebSocket-Protocol. Identity is exchanged either via a READY command (NULL) or as the first data message (no-mechanism), per RFC 45.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(ws, socket_type:, identity: "", as_server: false, mechanism: nil, max_message_size: nil, **opts) ⇒ Connection

Returns a new instance of Connection.

Parameters:

  • ws (#read, #send_binary, #flush, #close, #protocol) —

    WebSocket connection

  • socket_type (String) —

    our socket type (e.g. "REQ")

  • identity (String) (defaults to: "") —

    our identity

  • as_server (Boolean) (defaults to: false) —

    true on accepted side

  • mechanism (Object, nil) (defaults to: nil) —

    ignored — ZWS uses the negotiated subprotocol instead

  • max_message_size (Integer, nil) (defaults to: nil) —

    max frame body size

  • opts (Hash) —

    extra READY properties (NULL handshake only)



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/omq/transport/websocket/connection.rb', line 42

def initialize(ws, socket_type:, identity: "", as_server: false,
               mechanism: nil, max_message_size: nil, **opts)
  @ws               = ws
  @socket_type      = socket_type.to_s
  @identity         = identity || ""
  @as_server        = as_server
  @max_message_size = max_message_size
  @metadata         = opts.empty? ? nil : opts.transform_keys(&:to_s)

  @peer_socket_type = nil
  @peer_identity    = nil
  @peer_public_key  = nil
  @peer_properties  = nil

  # Advertise ZMTP 3.1 to upper layers so SUBSCRIBE/CANCEL always
  # go via send_command (the ZWS form per RFC 45) rather than the
  # legacy in-band byte-prefix form.
  @peer_major       = 3
  @peer_minor       = 1

  @last_received_at = nil
  @mutex            = Mutex.new
  @closed           = false
end

Instance Attribute Details

#last_received_at ⇒ Object (readonly)

Returns the value of attribute last_received_at.



30
31
32
 
# File 'lib/omq/transport/websocket/connection.rb', line 30

def last_received_at
  @last_received_at
end

#peer_identity ⇒ Object (readonly)

Returns the value of attribute peer_identity.



25
26
27
 
# File 'lib/omq/transport/websocket/connection.rb', line 25

def peer_identity
  @peer_identity
end

#peer_major ⇒ Object (readonly)

Returns the value of attribute peer_major.



28
29
30
 
# File 'lib/omq/transport/websocket/connection.rb', line 28

def peer_major
  @peer_major
end

#peer_minor ⇒ Object (readonly)

Returns the value of attribute peer_minor.



29
30
31
 
# File 'lib/omq/transport/websocket/connection.rb', line 29

def peer_minor
  @peer_minor
end

#peer_properties ⇒ Object (readonly)

Returns the value of attribute peer_properties.



27
28
29
 
# File 'lib/omq/transport/websocket/connection.rb', line 27

def peer_properties
  @peer_properties
end

#peer_public_key ⇒ Object (readonly)

Returns the value of attribute peer_public_key.



26
27
28
 
# File 'lib/omq/transport/websocket/connection.rb', line 26

def peer_public_key
  @peer_public_key
end

#peer_socket_type ⇒ Object (readonly)

Returns the value of attribute peer_socket_type.



24
25
26
 
# File 'lib/omq/transport/websocket/connection.rb', line 24

def peer_socket_type
  @peer_socket_type
end

#ws ⇒ Object (readonly)

Returns the value of attribute ws.



31
32
33
 
# File 'lib/omq/transport/websocket/connection.rb', line 31

def ws
  @ws
end

Instance Method Details

#close ⇒ Object 



221
222
223
224
225
226
227
 
# File 'lib/omq/transport/websocket/connection.rb', line 221

def close
  return if @closed
  @closed = true
  @ws.close
rescue IOError, ::Protocol::WebSocket::ClosedError
  # already closed
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


230
231
232
 
# File 'lib/omq/transport/websocket/connection.rb', line 230

def closed?
  @closed
end

#curve? ⇒ Boolean

Returns:

  • (Boolean) 


73
74
75
 
# File 'lib/omq/transport/websocket/connection.rb', line 73

def curve?
  false
end

#encrypted? ⇒ Boolean

Returns:

  • (Boolean) 


68
69
70
 
# File 'lib/omq/transport/websocket/connection.rb', line 68

def encrypted?
  false
end

#flush ⇒ Object 



121
122
123
124
125
 
# File 'lib/omq/transport/websocket/connection.rb', line 121

def flush
  @mutex.synchronize do
    @ws.flush
  end
end

#handshake! ⇒ void

This method returns an undefined value.

Performs the ZWS 2.0 mechanism handshake. Branches on the subprotocol negotiated during the HTTP upgrade.

Raises:

  • (Protocol::ZMTP::Error) —

    on protocol violation, unsupported subprotocol, or incompatible peer socket type



85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/omq/transport/websocket/connection.rb', line 85

def handshake!
  case @ws.protocol
  when "ZWS2.0/NULL"
    handshake_null!
    validate_peer_compatibility!
  when "ZWS2.0", nil
    handshake_no_mechanism!
    # No socket-type validation possible without READY exchange.
  else
    raise Protocol::ZMTP::Error, "unsupported ZWS subprotocol: #{@ws.protocol.inspect}"
  end
end

#heartbeat_expired?(timeout) ⇒ Boolean

Returns:

  • (Boolean) 


215
216
217
218
 
# File 'lib/omq/transport/websocket/connection.rb', line 215

def heartbeat_expired?(timeout)
  return false unless @last_received_at
  (Async::Clock.now - @last_received_at) > timeout
end

#read_frame ⇒ Protocol::ZMTP::Codec::Frame

Reads one ZWS frame. Auto-handles PING (replies PONG) and discards PONG. Returns a Protocol::ZMTP::Codec::Frame so upstream code (Subscription.parse, fan_out subscription listeners) sees the same shape as ZMTP/3.1.

Returns:

  • (Protocol::ZMTP::Codec::Frame)

Raises:

  • (EOFError) —

    on peer close



170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
 
# File 'lib/omq/transport/websocket/connection.rb', line 170

def read_frame
  loop do
    message = @ws.read
    unless message
      close
      raise EOFError, "ZWS connection closed"
    end

    bytes = message.buffer
    bytes = bytes.b if bytes.encoding != Encoding::BINARY
    flag, body = Codec.decode(bytes)

    if @max_message_size && body.bytesize > @max_message_size
      raise Protocol::ZMTP::Error,
            "ZWS frame exceeds max_message_size (#{body.bytesize} > #{@max_message_size})"
    end

    command = (flag & Codec::FLAG_COMMAND) != 0
    more    = !command && (flag & Codec::FLAG_MORE) != 0
    frame   = Protocol::ZMTP::Codec::Frame.new(body, more: more, command: command)

    touch_heartbeat

    if frame.command?
      cmd = Protocol::ZMTP::Codec::Command.from_body(frame.body)
      case cmd.name
      when "PING"
        _ttl, context = cmd.ping_ttl_and_context
        send_command(Protocol::ZMTP::Codec::Command.pong(context: context))
        next
      when "PONG"
        next
      end
    end

    return frame
  end
end

#receive_message ⇒ Array<String>

Reads a multi-frame message. Auto-handles PING/PONG and yields any other command frames to the caller block (matching Protocol::ZMTP::Connection#receive_message semantics).

Returns:

  • (Array<String>) —

    frame bodies

Raises:

  • (EOFError) —

    on peer close



143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/omq/transport/websocket/connection.rb', line 143

def receive_message
  frames = []

  loop do
    frame = read_frame

    if frame.command?
      yield frame if block_given?
      next
    end

    frames << frame.body
    break unless frame.more?
  end

  frames
end

#send_command(command) ⇒ Object 



128
129
130
131
132
133
 
# File 'lib/omq/transport/websocket/connection.rb', line 128

def send_command(command)
  @mutex.synchronize do
    @ws.send_binary(Codec.encode(command.to_body, command: true))
    @ws.flush
  end
end

#send_message(parts) ⇒ Object 



99
100
101
102
103
104
 
# File 'lib/omq/transport/websocket/connection.rb', line 99

def send_message(parts)
  @mutex.synchronize do
    write_frames(parts)
    @ws.flush
  end
end

#touch_heartbeat ⇒ Object 



210
211
212
 
# File 'lib/omq/transport/websocket/connection.rb', line 210

def touch_heartbeat
  @last_received_at = Async::Clock.now
end

#write_message(parts) ⇒ Object 



107
108
109
110
111
 
# File 'lib/omq/transport/websocket/connection.rb', line 107

def write_message(parts)
  @mutex.synchronize do
    write_frames(parts)
  end
end

#write_messages(messages) ⇒ Object 



114
115
116
117
118
 
# File 'lib/omq/transport/websocket/connection.rb', line 114

def write_messages(messages)
  @mutex.synchronize do
    messages.each { |parts| write_frames(parts) }
  end
end