Class: OMQ::Socket

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/omq/socket.rb

Overview

Socket base class.

Direct Known Subclasses

CHANNEL, CLIENT, DEALER, DISH, GATHER, PAIR, PEER, PUB, PULL, PUSH, RADIO, REP, REQ, ROUTER, SCATTER, SERVER, SUB, XPUB, XSUB

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(endpoints = nil) {|the| ... } ⇒ Socket

Use option accessors (e.g. socket.linger = 0) to configure post-construction.

Parameters:

  • endpoints (String, nil) (defaults to: nil) —

    optional endpoint with prefix convention (+@+ for bind, > for connect, plain uses subclass default)

Yield Parameters:

  • the (self) —

    socket, when a block is passed; the socket is #closed when the block returns (or raises).



79
80
 
# File 'lib/omq/socket.rb', line 79

def initialize(endpoints = nil)
end

Instance Attribute Details

#engine ⇒ Engine (readonly)

Returns the socket's engine. Exposed for peer tooling (omq-cli, omq-ractor) that needs to reach into the socket's internals — not part of the stable user API.

Returns:

  • (Engine) —

    the socket's engine. Exposed for peer tooling (omq-cli, omq-ractor) that needs to reach into the socket's internals — not part of the stable user API.



42
43
44
 
# File 'lib/omq/socket.rb', line 42

def engine
  @engine
end

#options ⇒ Options (readonly)

Returns:



35
36
37
 
# File 'lib/omq/socket.rb', line 35

def options
  @options
end

Class Method Details

.bind(endpoint, **opts) ⇒ Socket

Creates a new socket and binds it to the given endpoint.

Parameters:

  • endpoint (String)
  • opts (Hash) —

    keyword arguments forwarded to #initialize

Returns:



17
18
19
 
# File 'lib/omq/socket.rb', line 17

def self.bind(endpoint, **opts)
  new("@#{endpoint}", **opts)
end

.connect(endpoint, **opts) ⇒ Socket

Creates a new socket and connects it to the given endpoint.

Parameters:

  • endpoint (String)
  • opts (Hash) —

    keyword arguments forwarded to #initialize

Returns:



28
29
30
 
# File 'lib/omq/socket.rb', line 28

def self.connect(endpoint, **opts)
  new(">#{endpoint}", **opts)
end

Instance Method Details

#all_peers_gone ⇒ Async::Promise

Returns resolves when all peers disconnect (after having had peers).

Returns:

  • (Async::Promise) —

    resolves when all peers disconnect (after having had peers)



161
162
163
 
# File 'lib/omq/socket.rb', line 161

def all_peers_gone
  @engine.all_peers_gone
end

#attach_endpoints(endpoints, default:) ⇒ Object

Connects or binds based on endpoint prefix convention. Called from subclass initializers (including out-of-tree socket types).

Parameters:

  • endpoints (String, nil)
  • default (Symbol) —

    :connect or :bind



294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
 
# File 'lib/omq/socket.rb', line 294

def attach_endpoints(endpoints, default:)
  return unless endpoints

  if endpoints.is_a?(Array)
    endpoints.each { |ep| attach_endpoints(ep, default: default) }
    return
  end

  case endpoints
  when /\A@(.+)\z/
    bind($1)
  when /\A>(.+)\z/
    connect($1)
  else
    __send__(default, endpoints)
  end
end

#bind(endpoint, parent: nil, **opts) ⇒ URI::Generic

Binds to an endpoint.

Parameters:

  • endpoint (String)
  • parent (#async, nil) (defaults to: nil) —

    Async parent for the socket's task tree. Accepts any object that responds to #async — Async::Task, Async::Barrier, Async::Semaphore. When given, every task spawned under this socket (connection supervisors, reconnect loops, heartbeat, monitor) is placed under parent, so callers can coordinate teardown with their own Async tree. Only the first bind/connect call captures the parent — subsequent calls ignore the kwarg.

Returns:

  • (URI::Generic) —

    resolved endpoint URI (with auto-selected port for "tcp://host:0")



110
111
112
113
 
# File 'lib/omq/socket.rb', line 110

def bind(endpoint, parent: nil, **opts)
  ensure_parent_task(parent: parent)
  Reactor.run { @engine.bind(endpoint, **opts) } # TODO: use timeout?
end

#close ⇒ nil

Closes the socket and releases all resources. Drains pending sends up to linger seconds, then cascades teardown through the socket-level Async::Barrier — every connection's per-connection barrier is stopped, cancelling every pump.

Returns:

  • (nil) 


246
247
248
249
 
# File 'lib/omq/socket.rb', line 246

def close
  Reactor.run { @engine.close } # TODO: use timeout?
  nil
end

#close_read ⇒ void

This method returns an undefined value.

Signals end-of-stream on the receive side. A subsequent #receive call that would otherwise block returns nil.



177
178
179
 
# File 'lib/omq/socket.rb', line 177

def close_read
  @engine.dequeue_recv_sentinel
end

#closed? ⇒ Boolean

Returns true once the underlying engine is closed.

Returns:

  • (Boolean) —

    true once the underlying engine is closed.



253
254
255
 
# File 'lib/omq/socket.rb', line 253

def closed?
  @engine.closed?
end

#connect(endpoint, parent: nil, **opts) ⇒ URI::Generic

Connects to an endpoint.

Parameters:

  • endpoint (String)
  • parent (#async, nil) (defaults to: nil) —

    see #bind.

Returns:

  • (URI::Generic) —

    parsed endpoint URI



122
123
124
125
 
# File 'lib/omq/socket.rb', line 122

def connect(endpoint, parent: nil, **opts)
  ensure_parent_task(parent: parent)
  Reactor.run { @engine.connect(endpoint, **opts) } # TODO: use timeout?
end

#connection_count ⇒ Integer

Returns current number of peer connections.

Returns:

  • (Integer) —

    current number of peer connections



167
168
169
 
# File 'lib/omq/socket.rb', line 167

def connection_count
  @engine.connections.size
end

#disconnect(endpoint) ⇒ void

This method returns an undefined value.

Disconnects from an endpoint.

Parameters:

  • endpoint (String) 


133
134
135
 
# File 'lib/omq/socket.rb', line 133

def disconnect(endpoint)
  Reactor.run { @engine.disconnect(endpoint) } # TODO: use timeout?
end

#finalize_init ⇒ Object

Yields self if a block was given, then closes. Called by every subclass initializer so OMQ::PUSH.new { |p| ... } behaves like File.open: configure, use, auto-close.



87
88
89
90
91
92
93
94
 
# File 'lib/omq/socket.rb', line 87

def finalize_init
  return unless block_given?
  begin
    yield self
  ensure
    close
  end
end

#init_engine(socket_type, send_hwm: nil, recv_hwm: nil, send_timeout: nil, recv_timeout: nil, conflate: false, on_mute: nil, backend: nil) ⇒ Object

Initializes engine and options for a socket type. Called from subclass initializers (including out-of-tree socket types).

Parameters:

  • socket_type (Symbol)

Raises:

  • (ArgumentError) 


318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
 
# File 'lib/omq/socket.rb', line 318

def init_engine(socket_type, send_hwm: nil, recv_hwm: nil,
                send_timeout: nil, recv_timeout: nil, conflate: false,
                on_mute: nil, backend: nil)
  @options = Options.new
  @options.send_hwm     = send_hwm     if send_hwm
  @options.recv_hwm     = recv_hwm     if recv_hwm
  @options.send_timeout = send_timeout if send_timeout
  @options.recv_timeout = recv_timeout if recv_timeout
  @options.conflate     = conflate
  @options.on_mute      = on_mute      if on_mute
  backend_name = (backend || :ruby).to_sym
  if backend_name == :ruby && !Reactor.native_fiber_scheduler?
    raise NotImplementedError, "Ruby backend requires native Fiber.scheduler; use backend: :rust"
  end

  require_backend(backend_name) unless Backend.registered?(backend_name)

  engine_class = Backend.fetch(backend_name)
  raise ArgumentError, "unknown backend: #{backend}" unless engine_class

  @engine = engine_class.new(socket_type, @options)
end

#inspect ⇒ String

Returns:

  • (String) 


283
284
285
 
# File 'lib/omq/socket.rb', line 283

def inspect
  format("#<%s bound=%p>", self.class, @engine.listeners.keys)
end

#monitor(verbose: false) {|event| ... } ⇒ Async::Task

Yields lifecycle events for this socket.

Spawns a background fiber that reads from an internal event queue. The block receives MonitorEvent instances until the socket is closed or the returned task is stopped.

Examples:

task = socket.monitor do |event|
  case event
  in type: :connected, endpoint:
    puts "peer up: #{endpoint}"
  in type: :disconnected, endpoint:
    puts "peer down: #{endpoint}"
  end
end
# later:
task.stop

Yields:

  • (event) —

    called for each lifecycle event

Yield Parameters:

Returns:

  • (Async::Task) —

    the monitor task (call #stop to end early)



204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/omq/socket.rb', line 204

def monitor(verbose: false, &block)
  unless Reactor.native_fiber_scheduler?
    raise NotImplementedError, "Socket#monitor requires native Fiber.scheduler"
  end

  ensure_parent_task

  queue                   = Async::LimitedQueue.new(64)
  @engine.monitor_queue   = queue
  @engine.verbose_monitor = verbose

  Reactor.run do
    @engine.parent_task.async(transient: true, annotation: "monitor") do
      while (event = queue.dequeue)
        block.call(event)
      end
    rescue Async::Stop
    ensure
      @engine.monitor_queue = nil
      block.call(MonitorEvent.new(type: :monitor_stopped))
    end
  end
end

#peer_connected ⇒ Async::Promise

Returns resolves when first peer completes handshake.

Returns:

  • (Async::Promise) —

    resolves when first peer completes handshake



149
150
151
 
# File 'lib/omq/socket.rb', line 149

def peer_connected
  @engine.peer_connected
end

#reconnect_enabled=(val) ⇒ void

This method returns an undefined value.

Disable auto-reconnect for connected endpoints.

Parameters:

  • val (Boolean) 


234
235
236
 
# File 'lib/omq/socket.rb', line 234

def reconnect_enabled=(val)
  @engine.reconnect_enabled = val
end

#set_unbounded ⇒ nil

Set socket to use unbounded pipes (HWM=0).

Returns:

  • (nil) 


274
275
276
277
278
 
# File 'lib/omq/socket.rb', line 274

def set_unbounded
  @options.send_hwm = 0
  @options.recv_hwm = 0
  nil
end

#stop ⇒ nil

Immediate hard stop. Skips the linger drain and cascades teardown through the socket-level Async::Barrier. Intended for crash-path cleanup or when the caller already knows no pending sends matter.

Returns:

  • (nil) 


264
265
266
267
 
# File 'lib/omq/socket.rb', line 264

def stop
  Reactor.run { @engine.stop } # TODO: use timeout?
  nil
end

#subscriber_joined ⇒ Async::Promise

Returns resolves when first subscriber joins (PUB/XPUB only).

Returns:

  • (Async::Promise) —

    resolves when first subscriber joins (PUB/XPUB only)



155
156
157
 
# File 'lib/omq/socket.rb', line 155

def subscriber_joined
  @engine.subscriber_joined
end

#unbind(endpoint) ⇒ void

This method returns an undefined value.

Unbinds from an endpoint.

Parameters:

  • endpoint (String) 


143
144
145
 
# File 'lib/omq/socket.rb', line 143

def unbind(endpoint)
  Reactor.run { @engine.unbind(endpoint) } # TODO: use timeout?
end