Class: OMQ::RADIO

Inherits:
Socket show all
Includes:
Writable
Defined in:
lib/omq/radio_dish.rb

Overview

Group-based publisher socket (ZeroMQ RFC 48).

Sends messages to DISH peers that have joined the target group. Supports both TCP and UDP transports.

Constant Summary

Constants included from Writable

Writable::EMPTY_PART

Instance Attribute Summary

Attributes inherited from Socket

#engine, #options

Instance Method Summary collapse

Methods included from Writable

#wait_writable

Methods included from QueueWritable

#enqueue

Methods inherited from Socket

#all_peers_gone, #attach_endpoints, bind, #bind, #close, #close_read, connect, #connect, #connection_count, #disconnect, #finalize_init, #init_engine, #inspect, #monitor, #peer_connected, #reconnect_enabled=, #set_unbounded, #stop, #subscriber_joined, #unbind

Constructor Details

#initialize(endpoints = nil, linger: Float::INFINITY, on_mute: :drop_newest, conflate: false, backend: nil) ⇒ RADIO

Creates a new RADIO socket.

Parameters:

  • endpoints (String, Array<String>, nil) (defaults to: nil) —

    endpoint(s) to bind to

  • linger (Numeric) (defaults to: Float::INFINITY) —

    linger period in seconds (Float::INFINITY = wait forever, 0 = drop)

  • on_mute (Symbol) (defaults to: :drop_newest) —

    behaviour when HWM is reached (:drop_newest or :block)

  • conflate (Boolean) (defaults to: false) —

    if true, keep only the latest message per group per peer

  • backend (Object, nil) (defaults to: nil) —

    optional transport backend



31
32
33
34
35
 
# File 'lib/omq/radio_dish.rb', line 31

def initialize(endpoints = nil, linger: Float::INFINITY, on_mute: :drop_newest, conflate: false, backend: nil)
  init_engine(:RADIO, on_mute: on_mute, conflate: conflate, backend: backend)
  @options.linger = linger
  attach_endpoints(endpoints, default: :bind)
end

Instance Method Details

#<<(message) ⇒ self

Sends a message to a group via [group, body] array.

Parameters:

  • message (Array<String>) —
    group, body

Returns:

  • (self)

Raises:

  • (ArgumentError) 


70
71
72
73
 
# File 'lib/omq/radio_dish.rb', line 70

def <<(message)
  raise ArgumentError, "RADIO requires [group, body] array" unless message.is_a?(Array) && message.size == 2
  publish(message[0], message[1])
end

#publish(group, body) ⇒ self

Publishes a message to a group.

Parameters:

  • group (String) —

    group name

  • body (String) —

    message body

Returns:

  • (self) 


44
45
46
47
48
49
50
 
# File 'lib/omq/radio_dish.rb', line 44

def publish(group, body)
  parts = [group.b.freeze, body.b.freeze]
  Reactor.run timeout: @options.write_timeout do
    @engine.enqueue_send(parts)
  end
  self
end

#send(message, group: nil) ⇒ self

Sends a message to a group.

Parameters:

  • message (String) —

    message body (requires group: kwarg)

  • group (String) (defaults to: nil) —

    group name

Returns:

  • (self)

Raises:

  • (ArgumentError) 


59
60
61
62
 
# File 'lib/omq/radio_dish.rb', line 59

def send(message, group: nil)
  raise ArgumentError, "RADIO requires a group (use group: kwarg, publish, or << [group, body])" unless group
  publish(group, message)
end