Module: OMQ::QoS::SocketExt

Defined in:
lib/omq/qos/socket_ext.rb

Overview

Adds the qos accessors to Socket and validates qos= against the RFC. Fan-out socket types MUST refuse any QoS level above 0, and Integer levels are rejected in favour of OMQ::QoS instances (a 0.x → 0.3 break — see CHANGELOG).

Constant Summary collapse

FAN_OUT_TYPES = 
%i[PUB XPUB RADIO SUB XSUB DISH].freeze

Instance Method Summary collapse

Instance Method Details

#close ⇒ Object 



47
48
49
50
 
# File 'lib/omq/qos/socket_ext.rb', line 47

def close
  @options.qos&.shutdown
  super
end

#qos ⇒ Object 



16
17
18
 
# File 'lib/omq/qos/socket_ext.rb', line 16

def qos
  @options.qos
end

#qos=(value) ⇒ Object 



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/omq/qos/socket_ext.rb', line 21

def qos=(value)
  unless value.nil? || value.is_a?(OMQ::QoS)
    raise ArgumentError,
          "QoS must be nil (fire-and-forget) or an OMQ::QoS instance " \
          "(OMQ::QoS.at_least_once / .exactly_once / .exactly_once_and_processed); " \
          "received #{value.inspect}"
  end

  if value && FAN_OUT_TYPES.include?(@engine.socket_type)
    raise ArgumentError,
          "QoS > 0 is not supported for fan-out socket type #{@engine.socket_type}; " \
          "use a broker (e.g. Malamute) for reliable fan-out"
  end

  current = @options.qos
  if current && !current.equal?(value)
    raise ArgumentError,
          "OMQ::QoS already attached to this socket; a different instance " \
          "cannot be assigned (reset to nil first if supported)"
  end

  @options.qos = value
  value&.attach!(@engine)
end