Class: AceMQ::AMQP::Transport::Subscription

Inherits:

Object

• Object
• AceMQ::AMQP::Transport::Subscription

Defined in:

lib/acemq/amqp/transport.rb

Overview

A running subscription.

Stopping and closing are two steps rather than one because a handler that is still running needs its channel: closing it out from under an in-flight acknowledgement loses the message the handler just finished. Consumer#cancel stops delivery, waits, and only then closes.

Instance Method Summary

• #cancel ⇒ Object

  Both, for a caller with no handlers to wait for.

• #close ⇒ Object

  Releases the channel.

• #initialize(channel, consumer, position = nil) ⇒ Subscription constructor

  A new instance of Subscription.

• #open? ⇒ Boolean

  Whether the broker can still deliver down this subscription.

• #rewind_stream ⇒ Object private

  Moves a stream subscription's starting point to where it got to, so the re-subscribe a recovery is about to make carries on rather than starts over.

• #stop ⇒ Object

  Tells the broker to send no more.

Constructor Details

#initialize(channel, consumer, position = nil) ⇒ Subscription

Returns a new instance of Subscription.

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

def initialize(channel, consumer, position = nil)
  @channel = channel
  @consumer = consumer
  @position = position
end

Instance Method Details

#cancel ⇒ Object

Both, for a caller with no handlers to wait for.

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

def cancel
  stop
  close
end

#close ⇒ Object

Releases the channel. Nothing may be acknowledged after this.

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

def close
  @channel.close if @channel.open?
  nil
rescue StandardError
  nil
end

#open? ⇒ Boolean

Whether the broker can still deliver down this subscription.

The channel's own answer rather than a flag kept here. The two can disagree: a channel closed by the broker, or taken down by an error on it, stops delivery without anything in this process being told, and a health check reading a local flag would report a consumer that had been deaf for an hour as running.

Returns:

• (Boolean)

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

def open? = @channel.open?

#rewind_stream ⇒ Object

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.

Moves a stream subscription's starting point to where it got to, so the re-subscribe a recovery is about to make carries on rather than starts over. bunny 2.24 and 3.x both re-subscribe with the arguments the Bunny::Consumer carries, read when the re-subscribe is sent, so that hash is the one place a new offset can go. See StreamPosition.

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

def rewind_stream
  offset = @position&.resume
  @consumer.arguments[StreamPosition::ARGUMENT] = offset unless offset.nil?
end

#stop ⇒ Object

Tells the broker to send no more. Deliveries already handed over are unaffected.

Both steps check the channel first, and both are safe to call twice. A connection is closed by whatever gets there first — the consumer it owns, or the connection closing every consumer it tracks — and a basic.cancel on a channel that has already gone is a CHANNEL_ERROR, which RabbitMQ answers by dropping the whole connection. The symptom is a shutdown that takes ten seconds while bunny reconnects to a process that is exiting.

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

def stop
  @consumer.cancel if @channel.open?
  nil
rescue StandardError
  # Cancelling on a connection that has already gone is what shutdown
  # looks like when the broker went first, and raising here would turn
  # a tidy exit into a stack trace about a socket nobody is waiting on.
  nil
end