Class: AceMQ::AMQP::BatchPublish Private

Inherits:

Object

• Object
• AceMQ::AMQP::BatchPublish

Defined in:

lib/acemq/amqp/transport.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

One batch of messages, from the moment they are handed to a channel to the moment every one of them has an answer.

A class of its own because it is bookkeeping rather than transport: which delivery tag belongs to which payload, which basic.return belongs to which message, and what each of the two sets bunny leaves behind means. Every method on Transport is about one message and a channel; this is the only thing in the file that has to hold a whole batch in its head.

Instance Method Summary

• #initialize(messages, returns, confirms = ConfirmWait.new) ⇒ BatchPublish constructor private

  A new instance of BatchPublish.

• #run(channel, permits) ⇒ Array<String, PublishError> private

  Publishes every message in waves no wider than the permits allow, and answers for each of them.

Constructor Details

#initialize(messages, returns, confirms = ConfirmWait.new) ⇒ BatchPublish

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of BatchPublish.

Parameters:

• messages (Array<Hash>) —

  the keywords Transport#publish takes, one hash per message

• returns (ReturnedMessages) —

  the channel's returns, the same object a single mandatory publish reads

• confirms (ConfirmWait) (defaults to: ConfirmWait.new) —

  how to wait, so a lost connection ends the wait as it does for a single publish

# File 'lib/acemq/amqp/transport.rb', line 576

def initialize(messages, returns, confirms = ConfirmWait.new)
  @messages = messages
  @returns = returns
  @confirms = confirms
  @results = Array.new(messages.size)
  # Delivery tag to payload position, for the messages that are on the
  # wire right now with no answer back yet. Emptied by every wave.
  @wave = {}
  @waited = false
end

Instance Method Details

#run(channel, permits) ⇒ Array<String, PublishError>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Publishes every message in waves no wider than the permits allow, and answers for each of them.

A batch that fits inside the ceiling is one wave and behaves exactly as it always did: every message is written before anything is waited for, which is the point of the method. What changed is a batch that does not fit. That used to be written in full, however long the array was, with this process and the broker both holding all of it; it is now written in waves, each one confirmed — and its permits handed back — before the next is written. The results still come back in payload order, each message still gets its own answer, and a basic.return is still charged to the message whose id it carries.

Parameters:

• permits (PublishPermits) —

  the connection's ceiling, shared with every other publish on it

Returns:

• (Array<String, PublishError>) —

  one entry per message, in the order they were given

# File 'lib/acemq/amqp/transport.rb', line 604

def run(channel, permits)
  arm_returns(channel)
  @messages.each_index do |index|
    unless room_for_one(channel, permits)
      no_room_for(index, permits)
      return @results
    end

    publish_one(channel, index, permits)
  end
  settle(channel, permits) unless @wave.empty? && @waited
  @results
end