Module: OMQ::Transport::WebSocket

Defined in:
lib/omq/transport/websocket/transport.rb,
lib/omq/transport/websocket.rb,
lib/omq/transport/websocket/codec.rb,
lib/omq/transport/websocket/version.rb,
lib/omq/transport/websocket/connection.rb,
lib/omq/transport/websocket/options_ext.rb

Overview

ZeroMQ-over-WebSocket transport (ZWS 2.0, RFC 45). Registered for both ws:// and wss:// schemes. The only difference between the two is whether the underlying HTTP endpoint carries an SSL context (supplied via socket.tls_context=).

Defined Under Namespace

Modules: Codec, OptionsExt Classes: AcceptedConnection, Connection, Dialer, Error, Listener

Constant Summary collapse

VERSION = 
"0.1.0"

Class Method Summary collapse

Class Method Details

.connection_class ⇒ Class

Engine reads this in ConnectionLifecycle#handshake! to construct the per-connection codec/handshake driver. ZWS framing differs from ZMTP/3.1, so we substitute our own Connection class instead of Protocol::ZMTP::Connection.

Returns:

  • (Class) 


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

def connection_class
  OMQ::Transport::WebSocket::Connection
end

.dialer(endpoint, engine) ⇒ Dialer

Creates a WebSocket dialer.

Parameters:

  • endpoint (String)
  • engine (Engine)

Returns:



77
78
79
 
# File 'lib/omq/transport/websocket/transport.rb', line 77

def dialer(endpoint, engine, **)
  Dialer.new(endpoint, engine)
end

.listener(endpoint, engine) ⇒ Listener

Creates a bound WebSocket listener.

Parameters:

  • endpoint (String) —

    e.g. "ws://127.0.0.1:5555/" or "wss://host:443/zmq"

  • engine (Engine)

Returns:

Raises:

  • (Error) —

    if wss:// is used without a tls_context



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

def listener(endpoint, engine, **)
  scheme = endpoint[/\A([a-z]+):\/\//, 1]
  tls    = engine.options.tls_context

  if scheme == "wss" && tls.nil?
    raise Error, "wss:// bind requires options.tls_context"
  end

  subprotocols  = engine.options.ws_subprotocols
  path          = engine.options.ws_path
  http_endpoint = parse_http_endpoint(endpoint, tls)
  bound         = http_endpoint.bound
  actual_port   = bound.sockets.first.to_io.local_address.ip_port
  host          = http_endpoint.hostname
  host_part     = host.include?(":") ? "[#{host}]" : host
  shown         = "#{scheme}://#{host_part}:#{actual_port}#{path}"

  Listener.new(
    shown_endpoint: shown,
    bound:          bound,
    http_endpoint:  http_endpoint,
    subprotocols:   subprotocols,
    match_path:     path,
    engine:         engine,
    port:           actual_port,
  )
end

.parse_http_endpoint(endpoint, tls_context) ⇒ Async::HTTP::Endpoint

Parses endpoint into an Async::HTTP::Endpoint, attaching the SSL context when present (TLS for wss://).

Parameters:

  • endpoint (String)
  • tls_context (OpenSSL::SSL::SSLContext, nil)

Returns:

  • (Async::HTTP::Endpoint) 


101
102
103
104
105
106
107
 
# File 'lib/omq/transport/websocket/transport.rb', line 101

def parse_http_endpoint(endpoint, tls_context)
  if tls_context
    ::Async::HTTP::Endpoint.parse(endpoint, ssl_context: tls_context)
  else
    ::Async::HTTP::Endpoint.parse(endpoint)
  end
end

.validate_endpoint!(endpoint) ⇒ Object

Verifies that the endpoint URI is well-formed for HTTP/WS.

Parameters:

  • endpoint (String)

Raises:

  • (ArgumentError) 


87
88
89
90
91
 
# File 'lib/omq/transport/websocket/transport.rb', line 87

def validate_endpoint!(endpoint)
  uri = URI.parse(endpoint)
  raise ArgumentError, "missing host: #{endpoint.inspect}" unless uri.hostname
  raise ArgumentError, "missing port: #{endpoint.inspect}" unless uri.port
end