Class: AceMQ::AMQP::StreamPosition Private

Inherits:
Object
  • Object
show all
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.

How far a stream subscription has got, so a recovery carries on from there.

A queue forgets what it delivered, so re-subscribing after a lost connection takes whatever is left. A stream forgets nothing, and re-subscribing starts wherever x-stream-offset says -- and bunny re-subscribes with the arguments the subscription was first made with. A reader that began at first was handed the whole stream again (500 entries became 900 deliveries in the reproducer, on bunny 2.24 and 3.4 alike), and one that began at next skipped everything appended while it was away (50 of 200 never arrived).

So a re-subscribe starts at the oldest offset delivered and not yet settled, or one past the newest settled: what a queue would redeliver, and nothing it would not. Go's transport does the same.

Deliveries from before a recovery are counted no further once it has been worked out where to resume: the new subscription delivers them again, and a handler still finishing an old copy must not move the position past entries the new one has not reached.

Constant Summary collapse

ARGUMENT =

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

"x-stream-offset"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ StreamPosition

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 StreamPosition.



826
827
828
829
830
831
 
# File 'lib/acemq/amqp/transport.rb', line 826

def initialize
  @lock = Mutex.new
  @pending = Hash.new(0) # offset => deliveries not yet settled
  @settled = nil         # the newest offset settled
  @generation = 0
end

Class Method Details

.for(arguments) ⇒ 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.

One for a subscription that names a place in a stream, nil for any other.



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

def self.for(arguments) = (new if arguments.key?(ARGUMENT))

Instance Method Details

#resume ⇒ 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.

Where a re-subscribe starts, or nil when nothing was delivered and the subscription's own starting point still stands. Asking twice without a delivery in between gives the same answer, as a recovery that takes more than one attempt does.



861
862
863
864
865
866
867
868
869
870
871
 
# File 'lib/acemq/amqp/transport.rb', line 861

def resume
  @lock.synchronize do
    offset = @pending.empty? ? @settled&.+(1) : @pending.keys.min
    next nil if offset.nil?

    @pending.clear
    @settled = offset - 1
    @generation += 1
    offset
  end
end

#track(delivery) ⇒ 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.

The delivery, recorded as pending and settling itself on ack or nack. Recorded before the handler sees it and settled before the broker is told, so a recovery at any moment finds it either pending or done.



836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
 
# File 'lib/acemq/amqp/transport.rb', line 836

def track(delivery)
  offset = delivery.headers[ARGUMENT]
  return delivery unless offset.is_a?(Integer)

  generation = @lock.synchronize do
    @pending[offset] += 1
    @generation
  end
  ack = delivery.on_ack
  nack = delivery.on_nack
  delivery.on_ack = lambda do
    settle(offset, generation)
    ack&.call
  end
  delivery.on_nack = lambda do |requeue|
    settle(offset, generation)
    nack&.call(requeue)
  end
  delivery
end