Class: OMQ::Rust::Socket

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/rs/socket.rb

Overview

Base class for OMQ.rs-backed sockets.

Direct Known Subclasses

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(recv_timeout: nil, send_timeout: nil, curve_auth: nil, **options) ⇒ Socket

Creates a socket without binding or connecting it.

Parameters:

  • recv_timeout (Numeric, nil) (defaults to: nil) —

    receive timeout in seconds

  • send_timeout (Numeric, nil) (defaults to: nil) —

    send timeout in seconds

  • curve_auth (Array<String>, #call, nil) (defaults to: nil) —

    CURVE allowlist or authenticator

  • options (Hash) —

    native OMQ.rs socket options

Options Hash (**options):

  • :workload_profile (Symbol) —

    :throughput or :latency

  • :send_hwm (Integer) —

    outbound message capacity

  • :recv_hwm (Integer) —

    inbound message capacity

  • :recv_rate_limit (Hash) —

    per-connection :messages_per_second (or :rate) and :burst

  • :recv_ip_rate_limit (Hash) —

    per-IP :messages_per_second (or :rate) and :burst

  • :linger (Numeric) —

    close linger in seconds; positive infinity waits forever

  • :identity (String) —

    ZMTP socket identity

  • :router_mandatory (Boolean) —

    reject unroutable sends

  • :conflate (Boolean) —

    retain only latest received message

  • :heartbeat_interval (Numeric) —

    heartbeat period in seconds

  • :heartbeat_ttl (Numeric) —

    remote heartbeat TTL in seconds

  • :heartbeat_timeout (Numeric) —

    peer timeout in seconds

  • :handshake_timeout (Numeric) —

    handshake timeout in seconds

  • :max_pending_handshakes (Integer) —

    inbound handshake limit

  • :max_message_size (Integer) —

    maximum received message bytes

  • :sndbuf (Integer) —

    kernel send buffer bytes

  • :rcvbuf (Integer) —

    kernel receive buffer bytes

  • :large_message_threshold (Integer) —

    receive buffer threshold

  • :arena_threshold (Integer) —

    contiguous frame arena threshold

  • :transmit_slot_cap (Integer) —

    per-peer transmit bytes

  • :xpub_nodrop (Boolean) —

    block instead of dropping on mute

  • :reconnect_stop_conn_refused (Boolean) —

    stop refused reconnects

  • :on_mute (Symbol) —

    :block, :drop_newest, or :drop_oldest

  • :reconnect_interval (Numeric) —

    fixed reconnect delay in seconds

  • :reconnect_interval_min (Numeric) —

    minimum backoff in seconds

  • :reconnect_interval_max (Numeric) —

    maximum backoff in seconds

  • :compression_dict (String) —

    compression dictionary bytes

  • :compression_auto_train (Boolean) —

    train a zstd dictionary

  • :compression_threshold (Integer) —

    minimum bytes to compress

  • :compression_level (Integer) —

    zstd compression level

  • :compression_dict_capacity (Integer) —

    trained dictionary bytes

  • :max_recv_dict_size (Integer) —

    received dictionary byte limit

  • :compression_offload_threshold (Integer) —

    offload threshold; negative disables offloading

  • :mechanism_type (Symbol) —

    :null, :plain, or :curve

  • :plain_server (Boolean) —

    enable PLAIN server mode

  • :plain_username (String) —

    PLAIN client username

  • :plain_password (String) —

    PLAIN client password

  • :curve_server (Boolean) —

    enable CURVE server mode

  • :curve_publickey (String) —

    local raw or Z85 public key

  • :curve_secretkey (String) —

    local raw or Z85 secret key

  • :curve_serverkey (String) —

    CURVE server raw or Z85 public key

Raises:

  • (ArgumentError) —

    if an option is unknown or invalid



273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/omq/rs/socket.rb', line 273

def initialize(recv_timeout: nil, send_timeout: nil, curve_auth: nil, **options)
  socket_type = self.class.const_get(:SOCKET_TYPE, false)
  @socket_type = socket_type.to_s.downcase.to_sym
  unless SOCKET_TYPES.include?(@socket_type)
    raise ArgumentError, "unknown socket type: #{socket_type}"
  end

  @recv_timeout = recv_timeout
  @send_timeout = send_timeout
  @recv_batch   = []
  @request_waiting = false
  @reply_ready     = false
  @native       = Native::Socket.new(@socket_type.to_s.upcase)
  @native.set_options(normalize_options(options))
  @materialize_lock = Mutex.new
  @materialized = false
  @recv_io = nil
  @send_io = nil
  @peer_connected = false
  @subscriber_joined = false
  set_curve_auth(curve_auth) unless curve_auth.nil?
end

Instance Attribute Details

#socket_type ⇒ Symbol (readonly)

Returns lowercase socket pattern.

Returns:

  • (Symbol) —

    lowercase socket pattern



218
219
220
 
# File 'lib/omq/rs/socket.rb', line 218

def socket_type
  @socket_type
end

Instance Method Details

#bind(endpoint) ⇒ String

Binds socket to an endpoint.

Parameters:

  • endpoint (String, #to_str)

Returns:

  • (String) —

    resolved endpoint, including assigned ephemeral port



300
301
302
303
 
# File 'lib/omq/rs/socket.rb', line 300

def bind(endpoint)
  ensure_materialized
  @native.bind(String(endpoint))
end

#close ⇒ nil

Closes socket and releases native resources.

Returns:

  • (nil) 


625
626
627
628
629
630
631
632
 
# File 'lib/omq/rs/socket.rb', line 625

def close
  return if closed?

  @native.close
  close_wrapper(@recv_io)
  close_wrapper(@send_io)
  nil
end

#closed? ⇒ Boolean

Reports whether socket is closed.

Returns:

  • (Boolean) 


637
638
639
 
# File 'lib/omq/rs/socket.rb', line 637

def closed?
  @native.closed?
end

#connect(endpoint) ⇒ Socket

Connects socket to an endpoint.

Parameters:

  • endpoint (String, #to_str)

Returns:



309
310
311
312
313
 
# File 'lib/omq/rs/socket.rb', line 309

def connect(endpoint)
  ensure_materialized
  @native.connect(String(endpoint))
  self
end

#disconnect(endpoint) ⇒ Socket

Disconnects socket from an endpoint.

Parameters:

  • endpoint (String, #to_str)

Returns:



319
320
321
322
323
 
# File 'lib/omq/rs/socket.rb', line 319

def disconnect(endpoint)
  ensure_materialized
  @native.disconnect(String(endpoint))
  self
end

#each {|message| ... } ⇒ Enumerator?

Yields received messages until socket closes.

Yield Parameters:

  • message (Array<String, Integer>)

Returns:

  • (Enumerator, nil) —

    enumerator without a block



485
486
487
488
489
490
491
 
# File 'lib/omq/rs/socket.rb', line 485

def each
  return enum_for(__method__) unless block_given?

  loop { yield recv }
rescue IOError
  raise unless closed?
end

#join(group) ⇒ Socket

Joins DISH group.

Parameters:

  • group (String, #to_str)

Returns:



517
518
519
520
521
 
# File 'lib/omq/rs/socket.rb', line 517

def join(group)
  ensure_materialized
  @native.join(String(group).b)
  self
end

#leave(group) ⇒ Socket

Leaves DISH group.

Parameters:

  • group (String, #to_str)

Returns:



527
528
529
530
531
 
# File 'lib/omq/rs/socket.rb', line 527

def leave(group)
  ensure_materialized
  @native.leave(String(group).b)
  self
end

#monitor ⇒ Monitor

Returns socket lifecycle monitor.

Returns:



581
582
583
584
 
# File 'lib/omq/rs/socket.rb', line 581

def monitor
  ensure_materialized
  @monitor ||= Monitor.new(self)
end

#monitor_event(timeout: @recv_timeout) ⇒ Hash?

Receives next monitor event.

Parameters:

  • timeout (Numeric, nil) (defaults to: @recv_timeout) —

    maximum wait in seconds

Returns:

  • (Hash, nil)

Raises:

  • (IO::TimeoutError) —

    if timeout expires



601
602
603
604
605
606
607
608
609
610
 
# File 'lib/omq/rs/socket.rb', line 601

def monitor_event(timeout: @recv_timeout)
  return if closed?

  ensure_materialized
  event = @native.try_recv_monitor
  return event if event

  wait_for_native_fd(@native.monitor_fd, timeout, "monitor receive timed out")
  @native.try_recv_monitor
end

#monitor_fd ⇒ Integer

Returns monitor notification file descriptor.

Intended for event-loop adapters; use #monitor otherwise.

Returns:

  • (Integer) 


591
592
593
594
 
# File 'lib/omq/rs/socket.rb', line 591

def monitor_fd
  ensure_materialized
  @native.monitor_fd
end

#peer_info(routing_id) ⇒ Hash?

Returns metadata for live SERVER route.

Parameters:

  • routing_id (Integer) —

    SERVER routing ID

Returns:

  • (Hash, nil) —

    peer metadata, or nil for stale route

Raises:

  • (RuntimeError) —

    unless called on SERVER socket



340
341
342
343
 
# File 'lib/omq/rs/socket.rb', line 340

def peer_info(routing_id)
  ensure_materialized
  @native.peer_info(routing_id)
end

#publish(group, message) ⇒ Socket

Publishes RADIO message to group.

Parameters:

  • group (String, #to_str)
  • message (String, #to_str)

Returns:



538
539
540
 
# File 'lib/omq/rs/socket.rb', line 538

def publish(group, message)
  send(group, message)
end

#recv ⇒ Array<String, Integer> Also known as: receive

Receives next message.

Returns:

  • (Array<String, Integer>) —

    message frames; SERVER prepends routing ID

Raises:

  • (IO::TimeoutError) —

    if receive timeout expires

  • (IOError) —

    if socket closes



422
423
424
425
426
427
428
429
430
431
432
433
 
# File 'lib/omq/rs/socket.rb', line 422

def recv
  ensure_materialized
  message = try_recv
  return message if message

  loop do
    wait_for(@recv_io, @recv_timeout, "receive timed out")
    message = try_recv
    return message if message
    raise IOError, "socket closed" if closed?
  end
end

#send(message, *more) ⇒ Socket Also known as: <<

Sends message, blocking while send queue is full.

Parameters:

  • message (String, Integer, Array) —

    first frame or complete message

  • more (Array<String, Integer>) —

    additional frames

Returns:

Raises:

  • (IO::TimeoutError) —

    if send timeout expires

  • (ArgumentError, RuntimeError) —

    if message violates socket pattern



379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
 
# File 'lib/omq/rs/socket.rb', line 379

def send(message, *more)
  ensure_materialized
  parts = normalize_parts(message, more)
  validate_send_parts!(parts)
  validate_pattern_state_before_send!

  loop do
    result = enqueue(parts)
    if result == :ok
      sent!
      return self
    end

    wait_for(@send_io, @send_timeout, "send timed out")
    raise IOError, "socket closed" if closed?
  end
end

#set_curve_auth(authenticator = nil) {|peer| ... } ⇒ Socket

Configures CURVE client authorization on a server before materialization.

Parameters:

  • authenticator (Array<String>, #call, nil) (defaults to: nil) —

    public-key allowlist, callable receiving MechanismPeerInfo, or nil to allow valid clients

Yield Parameters:

Returns:

Raises:

  • (RuntimeError) —

    if socket is already materialized

  • (TypeError) —

    if authenticator is unsupported



353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
 
# File 'lib/omq/rs/socket.rb', line 353

def set_curve_auth(authenticator = nil, &block)
  raise RuntimeError, "CURVE authentication must be configured before bind or connect" if @materialized
  authenticator = block if block

  case authenticator
  when nil
    @native.clear_curve_auth
  when Array
    @native.set_curve_auth_keys(authenticator)
  else
    unless authenticator.respond_to?(:call)
      raise TypeError, "CURVE authenticator must be an Array, callable, or nil"
    end

    @native.set_curve_auth_callback(Rust.wrap_curve_authenticator(authenticator))
  end
  self
end

#subscribe(prefix = "") ⇒ Socket

Adds SUB or XSUB subscription prefix.

Parameters:

  • prefix (String, #to_str) (defaults to: "")

Returns:



497
498
499
500
501
 
# File 'lib/omq/rs/socket.rb', line 497

def subscribe(prefix = "")
  ensure_materialized
  @native.subscribe(String(prefix).b)
  self
end

#try_monitor_event ⇒ Hash?

Attempts to receive monitor event without blocking.

Returns:

  • (Hash, nil) 


615
616
617
618
619
620
 
# File 'lib/omq/rs/socket.rb', line 615

def try_monitor_event
  return if closed?

  ensure_materialized
  @native.try_recv_monitor
end

#try_recv ⇒ Array<String, Integer>?

Attempts to receive without blocking.

Returns:

  • (Array<String, Integer>, nil) —

    next message, or nil if none is ready



441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
 
# File 'lib/omq/rs/socket.rb', line 441

def try_recv
  ensure_materialized
  unless @recv_batch.empty?
    message = @recv_batch.shift
    received!
    return message
  end

  message = if ROUTED_TYPES.include?(@socket_type)
    @native.try_recv_routed
  elsif (batch = @native.try_recv_batch)
    first = batch.shift
    @recv_batch = batch
    first
  end
  received! if message
  message
end

#try_send(message, *more) ⇒ Boolean

Attempts to send without blocking.

Parameters:

  • message (String, Integer, Array) —

    first frame or complete message

  • more (Array<String, Integer>) —

    additional frames

Returns:

  • (Boolean) —

    whether message was queued

Raises:

  • (ArgumentError, RuntimeError) —

    if message violates socket pattern



406
407
408
409
410
411
412
413
414
415
 
# File 'lib/omq/rs/socket.rb', line 406

def try_send(message, *more)
  ensure_materialized
  parts = normalize_parts(message, more)
  validate_send_parts!(parts)
  validate_pattern_state_before_send!
  return false unless enqueue(parts) == :ok

  sent!
  true
end

#unbind(endpoint) ⇒ Socket

Stops listening on an endpoint.

Parameters:

  • endpoint (String, #to_str)

Returns:



329
330
331
332
333
 
# File 'lib/omq/rs/socket.rb', line 329

def unbind(endpoint)
  ensure_materialized
  @native.unbind(String(endpoint))
  self
end

#unsubscribe(prefix = "") ⇒ Socket

Removes SUB or XSUB subscription prefix.

Parameters:

  • prefix (String, #to_str) (defaults to: "")

Returns:



507
508
509
510
511
 
# File 'lib/omq/rs/socket.rb', line 507

def unsubscribe(prefix = "")
  ensure_materialized
  @native.unsubscribe(String(prefix).b)
  self
end

#wait_for_peer(timeout: nil) ⇒ Socket

Waits until first peer completes handshake.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    maximum wait in seconds

Returns:

Raises:

  • (IO::TimeoutError) —

    if timeout expires

  • (IOError) —

    if socket closes first



548
549
550
551
552
553
554
555
556
557
558
 
# File 'lib/omq/rs/socket.rb', line 548

def wait_for_peer(timeout: nil)
  raise IOError, "socket closed" if closed?
  return self if @peer_connected

  ensure_materialized
  wait_for_native_fd(@native.peer_connected_fd, timeout, "peer connection timed out")
  raise IOError, "socket closed" if closed?

  @peer_connected = true
  self
end

#wait_for_subscriber(timeout: nil) ⇒ Socket

Waits until PUB, XPUB, or RADIO receives first subscription.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    maximum wait in seconds

Returns:

Raises:

  • (IO::TimeoutError) —

    if timeout expires

  • (IOError) —

    if socket closes first



566
567
568
569
570
571
572
573
574
575
576
 
# File 'lib/omq/rs/socket.rb', line 566

def wait_for_subscriber(timeout: nil)
  raise IOError, "socket closed" if closed?
  return self if @subscriber_joined

  ensure_materialized
  wait_for_native_fd(@native.subscriber_joined_fd, timeout, "subscriber timed out")
  raise IOError, "socket closed" if closed?

  @subscriber_joined = true
  self
end

#wait_readable(timeout: @recv_timeout) ⇒ true

Waits for receive notification.

Notification may represent a message, close, or explicit #wake_recv.

Parameters:

  • timeout (Numeric, nil) (defaults to: @recv_timeout) —

    maximum wait in seconds

Returns:

  • (true)

Raises:

  • (IO::TimeoutError) —

    if timeout expires



467
468
469
470
471
 
# File 'lib/omq/rs/socket.rb', line 467

def wait_readable(timeout: @recv_timeout)
  ensure_materialized
  wait_for(@recv_io, timeout, "receive timed out")
  true
end

#wake_recv ⇒ Socket

Wakes a thread or fiber blocked in #wait_readable.

Returns:



476
477
478
479
 
# File 'lib/omq/rs/socket.rb', line 476

def wake_recv
  @native.wake_recv if @materialized
  self
end