Class: AceMQ::AMQP::Patterns::InMemoryOutboxStore

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/patterns/outbox.rb

Overview

An outbox in this process.

It has none of the property the pattern exists for: nothing here shares a transaction with anybody's database, so a crash between the work committing and the record being written loses the message exactly as publishing directly would. It is for tests, and for seeing the shape of the thing before wiring the real one.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ InMemoryOutboxStore

Returns a new instance of InMemoryOutboxStore.



70
71
72
73
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 70

def initialize
  @records = {}
  @lock = Mutex.new
end

Instance Method Details

#add(record) ⇒ Object

Records a message to be published.

Adding the same record twice is not an error — the caller may be retrying its own transaction — but it must not become two messages.

Raises:

  • (ArgumentError) 


79
80
81
82
83
84
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 79

def add(record)
  raise ArgumentError, "an outbox record needs an id" if record.id.to_s.empty?

  @lock.synchronize { @records[record.id] ||= record }
  nil
end

#mark_published(id) ⇒ Object

Removes a record the broker has confirmed.



98
99
100
101
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 98

def mark_published(id)
  @lock.synchronize { @records.delete(id) }
  nil
end

#pending(limit = 0) ⇒ Object

The records waiting, oldest first.

Oldest first because the order they were written in is usually the order the writer meant, and a relay that publishes them out of order has invented a reordering nobody asked for.



91
92
93
94
95
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 91

def pending(limit = 0)
  waiting = @lock.synchronize { @records.values }
                 .sort_by { |record| [record.created_at, record.id] }
  limit.positive? ? waiting.first(limit) : waiting
end

#size ⇒ Object

How many records are waiting.



104
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 104

def size = @lock.synchronize { @records.size }