Module: OMQ::QoS::WritableExt

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

Overview

Prepended onto Writable so that at QoS >= 2 #send returns an Async::Promise resolving to :delivered or a DeadLetter. At QoS 0/1 the original Writable#send is preserved (returns self).

The Promise is stashed on the socket's OMQ::QoS via #enqueue_promise keyed by parts.object_id. The routing layer (RoundRobinExt#write_batch) takes it back out when the message reaches the wire and attaches it to the PeerRegistry::Entry. This keeps the send queue's item type uniform (Array) across QoS levels, so batch_bytes, emit_verbose_msg_sent, and RecvPump need no awareness of QoS.

Instance Method Summary collapse

Instance Method Details

#send(message) ⇒ Object

Raises:

  • (ArgumentError) 


139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/omq/qos/socket_ext.rb', line 139

def send(message)
  qos = @options.qos
  return super if qos.nil? || qos.level < 2

  parts = message.is_a?(Array) ? message : [message]
  raise ArgumentError, "message has no parts" if parts.empty?

  parts = parts.map(&:to_str) if parts.any? { |part| !part.is_a?(String) }
  parts.each do |part|
    part.force_encoding(Encoding::BINARY) unless part.frozen? || part.encoding == Encoding::BINARY
    part.freeze
  end
  parts.freeze

  promise = Async::Promise.new
  qos.enqueue_promise(parts, promise)

  if @engine.on_io_thread?
    OMQ::Reactor.run(timeout: @options.write_timeout) { @engine.enqueue_send(parts) }
  elsif (timeout = @options.write_timeout)
    Async::Task.current.with_timeout(timeout, IO::TimeoutError) { @engine.enqueue_send(parts) }
  else
    @engine.enqueue_send(parts)
  end

  promise
end