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

Raises:

  • (ArgumentError) —

    if an option is invalid



161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/omq/rs/socket.rb', line 161

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
  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



151
152
153
 
# File 'lib/omq/rs/socket.rb', line 151

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



186
187
188
189
 
# File 'lib/omq/rs/socket.rb', line 186

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

#close ⇒ nil

Closes socket and releases native resources.

Returns:

  • (nil) 


493
494
495
496
497
498
499
500
 
# File 'lib/omq/rs/socket.rb', line 493

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) 


505
506
507
 
# File 'lib/omq/rs/socket.rb', line 505

def closed?
  @native.closed?
end

#connect(endpoint) ⇒ Socket

Connects socket to an endpoint.

Parameters:

  • endpoint (String, #to_str)

Returns:



195
196
197
198
199
 
# File 'lib/omq/rs/socket.rb', line 195

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:



205
206
207
208
209
 
# File 'lib/omq/rs/socket.rb', line 205

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



371
372
373
374
375
376
377
 
# File 'lib/omq/rs/socket.rb', line 371

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:



403
404
405
406
407
 
# File 'lib/omq/rs/socket.rb', line 403

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

#leave(group) ⇒ Socket

Leaves DISH group.

Parameters:

  • group (String, #to_str)

Returns:



413
414
415
416
417
 
# File 'lib/omq/rs/socket.rb', line 413

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

#monitor ⇒ Monitor

Returns socket lifecycle monitor.

Returns:



453
454
455
456
 
# File 'lib/omq/rs/socket.rb', line 453

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



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

def monitor_event(timeout: @recv_timeout)
  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) 


463
464
465
466
 
# File 'lib/omq/rs/socket.rb', line 463

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



226
227
228
229
 
# File 'lib/omq/rs/socket.rb', line 226

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:



424
425
426
 
# File 'lib/omq/rs/socket.rb', line 424

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



308
309
310
311
312
313
314
315
316
317
318
319
 
# File 'lib/omq/rs/socket.rb', line 308

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



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/omq/rs/socket.rb', line 265

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 authentication before socket 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



239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/omq/rs/socket.rb', line 239

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:



383
384
385
386
387
 
# File 'lib/omq/rs/socket.rb', line 383

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) 


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

def try_monitor_event
  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



327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
 
# File 'lib/omq/rs/socket.rb', line 327

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



292
293
294
295
296
297
298
299
300
301
 
# File 'lib/omq/rs/socket.rb', line 292

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:



215
216
217
218
219
 
# File 'lib/omq/rs/socket.rb', line 215

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:



393
394
395
396
397
 
# File 'lib/omq/rs/socket.rb', line 393

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



433
434
435
436
437
 
# File 'lib/omq/rs/socket.rb', line 433

def wait_for_peer(timeout: nil)
  ensure_materialized
  wait_for_native_fd(@native.peer_connected_fd, timeout, "peer connection timed out")
  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



444
445
446
447
448
 
# File 'lib/omq/rs/socket.rb', line 444

def wait_for_subscriber(timeout: nil)
  ensure_materialized
  wait_for_native_fd(@native.subscriber_joined_fd, timeout, "subscriber timed out")
  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



353
354
355
356
357
 
# File 'lib/omq/rs/socket.rb', line 353

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:



362
363
364
365
 
# File 'lib/omq/rs/socket.rb', line 362

def wake_recv
  @native.wake_recv if @materialized
  self
end