Module: OMQ::CLI::SocketSetup

Defined in:
lib/omq/cli/socket_setup.rb

Overview

Stateless helper for socket construction and configuration. All methods are module-level so callers compose rather than inherit.

Constant Summary collapse

DEFAULT_HWM =

Default high water mark applied when the user does not pass --hwm. Lower than libzmq's default (1000) to keep memory footprint bounded for the typical CLI use cases (interactive debugging, short-lived pipelines). 64 matches the recv pump's per-fairness-batch limit (one batch exactly fills a full queue).

64
DEFAULT_RECV_MAXSZ =

Default max inbound message size applied when the user does not pass --recv-maxsz. The omq library itself is unlimited by default; the CLI caps inbound messages at 1 MiB so that a misconfigured or malicious peer cannot force arbitrary memory allocation on a terminal user. Users can raise it with --recv-maxsz N, or disable it entirely with --recv-maxsz 0.

1 << 20
DEFAULT_ZSTD_LEVEL =

Default zstd level when the user picks zstd without specifying one (e.g. --compress=zstd). Matches the -z shortcut.

-3

Class Method Summary collapse

Class Method Details

.apply_options(sock, config) ⇒ Object

Apply common socket options from config to sock.



116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/omq/cli/socket_setup.rb', line 116

def self.apply_options(sock, config)
  sock.linger             = config.linger
  sock.recv_timeout       = config.timeout       if config.timeout
  sock.send_timeout       = config.timeout       if config.timeout
  sock.reconnect_interval = config.reconnect_ivl if config.reconnect_ivl
  sock.heartbeat_interval = config.heartbeat_ivl if config.heartbeat_ivl
  # nil → default; 0 stays 0 (unbounded), any other integer is taken as-is.
  sock.send_hwm           = config.send_hwm || DEFAULT_HWM
  sock.recv_hwm           = config.recv_hwm || DEFAULT_HWM
  sock.sndbuf             = config.sndbuf        if config.sndbuf
  sock.rcvbuf             = config.rcvbuf        if config.rcvbuf
end

.apply_recv_maxsz(sock, config) ⇒ Object

--recv-maxsz: nil → 1 MiB default; 0 → explicitly unlimited; else → as-is.



144
145
146
147
148
149
150
151
 
# File 'lib/omq/cli/socket_setup.rb', line 144

def self.apply_recv_maxsz(sock, config)
  sock.max_message_size =
    case config.recv_maxsz
    when nil then DEFAULT_RECV_MAXSZ
    when 0   then nil
    else          config.recv_maxsz
    end
end

.attach(sock, config, verbose: 0, timestamps: nil) ⇒ Object

Bind/connect sock using URL strings from config.binds / config.connects. verbose gates logging (>= 1), timestamps controls prefix.



157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/omq/cli/socket_setup.rb', line 157

def self.attach(sock, config, verbose: 0, timestamps: nil)
  opts = compress_opts(config)

  config.binds.each do |url|
    uri = sock.bind(compress_url(url, config), **opts)
    CLI::Term.write_attach(:bind, uri.to_s, timestamps) if verbose >= 1
  end

  config.connects.each do |url|
    sock.connect(compress_url(url, config), **opts)
    CLI::Term.write_attach(:connect, url, timestamps) if verbose >= 1
  end
end

.attach_endpoints(sock, endpoints, config: nil, verbose: 0, timestamps: nil, side: nil) ⇒ Object

Bind/connect sock from an Array of Endpoint objects. Used by PipeRunner, which works with structured endpoint lists. verbose gates logging (>= 1), timestamps controls prefix. side selects per-socket compression (:in for PULL, :out for PUSH in a pipe); nil falls back to global compress settings.



178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/omq/cli/socket_setup.rb', line 178

def self.attach_endpoints(sock, endpoints, config: nil, verbose: 0, timestamps: nil, side: nil)
  opts = config ? compress_opts(config, side: side) : {}

  endpoints.each do |ep|
    url = config ? compress_url(ep.url, config, side: side) : ep.url

    if ep.bind?
      uri = sock.bind(url, **opts)
      CLI::Term.write_attach(:bind, uri.to_s, timestamps) if verbose >= 1
    else
      sock.connect(url, **opts)
      CLI::Term.write_attach(:connect, ep.url, timestamps) if verbose >= 1
    end
  end
end

.backend_kwargs(config) ⇒ Object

Returns socket constructor kwargs for the selected backend.



91
92
93
94
95
96
 
# File 'lib/omq/cli/socket_setup.rb', line 91

def self.backend_kwargs(config)
  backend = backend_name(config)
  return {} if backend.nil? || backend == :ruby

  { backend: backend }
end

.backend_name(config) ⇒ Object

Returns the effective backend name after legacy --ffi handling.



101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/omq/cli/socket_setup.rb', line 101

def self.backend_name(config)
  backend = if config.respond_to?(:ffi) && config.ffi
              :libzmq
            elsif config.respond_to?(:backend)
              config.backend
            else
              :ruby
            end
  backend = :libzmq if backend == :ffi
  backend || :ruby
end

.build(klass, config) ⇒ Object

Create and fully configure a socket from klass and config.



132
133
134
135
136
137
138
139
140
 
# File 'lib/omq/cli/socket_setup.rb', line 132

def self.build(klass, config)
  sock = klass.new(**backend_kwargs(config))
  sock.conflate = true if config.conflate && %w[pub radio].include?(config.type_name)
  apply_options(sock, config)
  apply_recv_maxsz(sock, config)
  sock.identity         = config.identity   if config.identity
  sock.router_mandatory = true if config.type_name == "router"
  sock
end

.compress_flag(config, side: nil) ⇒ Object

Human-readable CLI flag for a resolved compress state, suitable for process titles and proctitle-like log lines.



80
81
82
83
84
85
86
 
# File 'lib/omq/cli/socket_setup.rb', line 80

def self.compress_flag(config, side: nil)
  codec, level = resolve_compress(config, side)
  case codec
  when :zstd then level == 3 ? "-Z" : "-z"
  when :lz4  then "--lz4"
  end
end

.compress_opts(config, side: nil) ⇒ Object

Returns bind/connect kwargs for the chosen codec on the given side. Zstd accepts a level:; LZ4 takes no tuning knobs.



68
69
70
71
72
73
74
 
# File 'lib/omq/cli/socket_setup.rb', line 68

def self.compress_opts(config, side: nil)
  codec, level = resolve_compress(config, side)
  case codec
  when :zstd then { level: level || DEFAULT_ZSTD_LEVEL }
  else            {}
  end
end

.compress_url(url, config, side: nil) ⇒ Object

Upgrades a tcp:// URL to the codec-specific variant (+zstd+tcp://+ or +lz4+tcp://+) when compression is enabled for the given side. Returns the URL unchanged for non-TCP or when compression is off.



56
57
58
59
60
61
62
 
# File 'lib/omq/cli/socket_setup.rb', line 56

def self.compress_url(url, config, side: nil)
  codec, _ = resolve_compress(config, side)
  return url if codec.nil?
  return url unless url.start_with?("tcp://")

  url.sub("tcp://", "#{codec}+tcp://")
end

.kill_on_protocol_error(sock, event) ⇒ Object

CLI-level policy: a peer that commits a protocol-level violation (Protocol::ZMTP::Error — oversized frame, decompression bytebomb, bad framing, …) is almost certainly a misconfiguration the user needs to see. Mark sock dead so the next receive raises SocketDeadError. The library itself just drops the connection and keeps serving the others; this stricter policy is CLI-only.

Parameters:

  • sock (OMQ::Socket)
  • event (OMQ::MonitorEvent) 


257
258
259
260
261
262
 
# File 'lib/omq/cli/socket_setup.rb', line 257

def self.kill_on_protocol_error(sock, event)
  return unless event.type == :disconnected
  error = event.detail && event.detail[:error]
  return unless error.is_a?(Protocol::ZMTP::Error)
  sock.engine.signal_fatal_error(error)
end

.resolve_compress(config, side) ⇒ Object

Resolves compression state for a given pipe side (:in, :out, or nil). Per-side settings (in_compress / out_compress) override the global config.compress. Returns [codec, level] where codec is one of nil (off), :zstd, or :lz4.



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/omq/cli/socket_setup.rb', line 34

def self.resolve_compress(config, side)
  scoped =
    case side
    when :in  then !config.in_compress.nil?
    when :out then !config.out_compress.nil?
    end

  if scoped && side == :in
    [config.in_compress,  config.in_compress_level]
  elsif scoped && side == :out
    [config.out_compress, config.out_compress_level]
  else
    [config.compress, config.compress_level]
  end
end

.setup_curve(sock, config) ⇒ Object

Configure CURVE encryption on sock using config and env vars.



210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
 
# File 'lib/omq/cli/socket_setup.rb', line 210

def self.setup_curve(sock, config)
  server_key_z85 = config.curve_server_key || ENV["OMQ_SERVER_KEY"]
  server_mode    = config.curve_server || (ENV["OMQ_SERVER_PUBLIC"] && ENV["OMQ_SERVER_SECRET"])

  return unless server_key_z85 || server_mode

  crypto = CLI.load_curve_crypto(config.crypto || ENV["OMQ_CRYPTO"], verbose: config.verbose >= 1)
  require "protocol/zmtp/mechanism/curve"

  if server_key_z85
    server_key = Protocol::ZMTP::Z85.decode(server_key_z85)
    client_key = crypto::PrivateKey.generate
    sock.mechanism = Protocol::ZMTP::Mechanism::Curve.client(
      public_key: client_key.public_key.to_s,
      secret_key: client_key.to_s,
      server_key: server_key,
      crypto: crypto
    )
  elsif server_mode
    if ENV["OMQ_SERVER_PUBLIC"] && ENV["OMQ_SERVER_SECRET"]
      server_pub = Protocol::ZMTP::Z85.decode(ENV["OMQ_SERVER_PUBLIC"])
      server_sec = Protocol::ZMTP::Z85.decode(ENV["OMQ_SERVER_SECRET"])
    else
      key        = crypto::PrivateKey.generate
      server_pub = key.public_key.to_s
      server_sec = key.to_s
    end
    sock.mechanism = Protocol::ZMTP::Mechanism::Curve.server(
      public_key: server_pub,
      secret_key: server_sec,
      crypto: crypto
    )
    $stderr.puts "OMQ_SERVER_KEY='#{Protocol::ZMTP::Z85.encode(server_pub)}'"
  end
end

.setup_subscriptions(sock, config) ⇒ Object

Subscribe or join groups on sock according to config.



197
198
199
200
201
202
203
204
205
 
# File 'lib/omq/cli/socket_setup.rb', line 197

def self.setup_subscriptions(sock, config)
  case config.type_name
  when "sub"
    prefixes = config.subscribes.empty? ? [""] : config.subscribes
    prefixes.each { |p| sock.subscribe(p) }
  when "dish"
    config.joins.each { |g| sock.join(g) }
  end
end