Module: NNQ::CLI::SocketSetup

Defined in:
lib/nnq/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. 64 matches the send pump’s per-fairness-batch limit (one batch exactly fills a full queue).

64
DEFAULT_RECV_MAXSZ =

Default max inbound message size (1 MiB) so a misconfigured or malicious peer can’t 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

Class Method Summary collapse

Class Method Details

.apply_options(sock, config) ⇒ Object

Apply post-construction socket options from config to sock. send_hwm and linger are construction-time kwargs (see build); the rest of the options are set here and read later by the engine/transports.



25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/nnq/cli/socket_setup.rb', line 25

def self.apply_options(sock, config)
  sock.options.read_timeout       = config.timeout       if config.timeout
  sock.options.write_timeout      = config.timeout       if config.timeout
  sock.options.reconnect_interval = config.reconnect_ivl if config.reconnect_ivl

  case config.recv_maxsz
  when nil
    sock.options.max_message_size = DEFAULT_RECV_MAXSZ
  when 0  
    sock.options.max_message_size = nil
  else
    config.recv_maxsz
  end
end

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

Bind/connect sock using URL strings from config.binds / config.connects. verbose is the integer verbosity level (0 = silent).



57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/nnq/cli/socket_setup.rb', line 57

def self.attach(sock, config, verbose: 0)
  config.binds.each do |url|
    sock.bind(url)
    CLI::Term.write_attach(:bind, sock.last_endpoint, verbose) if verbose >= 1
  end

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

.attach_endpoints(sock, endpoints, verbose: 0) ⇒ Object

Bind/connect sock from an Array of Endpoint objects. Used by PipeRunner, which works with structured endpoint lists. verbose is the integer verbosity level (0 = silent).



73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/nnq/cli/socket_setup.rb', line 73

def self.attach_endpoints(sock, endpoints, verbose: 0)
  endpoints.each do |ep|
    if ep.bind?
      sock.bind(ep.url)
      CLI::Term.write_attach(:bind, sock.last_endpoint, verbose) if verbose >= 1
    else
      sock.connect(ep.url)
      CLI::Term.write_attach(:connect, ep.url, verbose) if verbose >= 1
    end
  end
end

.build(klass, config) ⇒ Object

Create and fully configure a socket from klass and config. nnq’s Socket constructor takes linger + send_hwm directly (send_hwm is captured during routing init and can’t be changed after the fact), so we pass them there.



45
46
47
48
49
50
51
52
 
# File 'lib/nnq/cli/socket_setup.rb', line 45

def self.build(klass, config)
  linger   = config.linger
  send_hwm = config.send_hwm || DEFAULT_HWM
  sock     = klass.new(linger:, send_hwm:)

  apply_options(sock, config)
  sock
end

.maybe_wrap_zstd(sock, mode) ⇒ Object

Wrap sock with NNQ::Zstd if mode is :fast or :balanced. Returns the wrapper (decorator) or sock unchanged.



88
89
90
91
92
93
 
# File 'lib/nnq/cli/socket_setup.rb', line 88

def self.maybe_wrap_zstd(sock, mode)
  return sock unless mode
  require "nnq/zstd"
  level = mode == :balanced ? 3 : -3
  NNQ::Zstd.wrap(sock, level: level)
end

.setup_subscriptions(sock, config) ⇒ Object

Subscribe to prefixes on a SUB socket.

Unlike ZeroMQ, nng’s sub0 starts with an empty subscription set, meaning no messages match. If the user passed no ‘–subscribe` flags, default to subscribing to the empty prefix so the CLI feels like `nngcat` / `omq sub`: receive everything by default.



102
103
104
105
106
 
# File 'lib/nnq/cli/socket_setup.rb', line 102

def self.setup_subscriptions(sock, config)
  return unless config.type_name == "sub"
  prefixes = config.subscribes.empty? ? [""] : config.subscribes
  prefixes.each { |p| sock.subscribe(p) }
end