Class: AceMQ::AMQP::Patterns::OutboxRelay

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

Overview

Publishes what the outbox holds.

relay = Patterns::OutboxRelay.new(mq, store, interval: 1)
relay.start
at_exit { relay.close }

Deliberately at-least-once. A record is removed only once the broker has confirmed the message, so a crash in between publishes it again — which is why consumers of anything sent this way have to be idempotent, and why idempotent is in the same library. The alternative, removing first, loses messages instead, and a lost message is the worse of the two: a duplicate can be recognised, an absence cannot.

Constant Summary collapse

DEFAULT_INTERVAL =

How often the outbox is swept, and how many records go per sweep.

1.0
DEFAULT_BATCH = 
100

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, store, interval: DEFAULT_INTERVAL, batch: DEFAULT_BATCH, on_error: nil) ⇒ OutboxRelay

Returns a new instance of OutboxRelay.

Parameters:

  • connection (Connection, Transport) —

    where to publish

  • store (OutboxStore) —

    what to publish

  • interval (Numeric) (defaults to: DEFAULT_INTERVAL) —

    seconds between sweeps

  • batch (Integer) (defaults to: DEFAULT_BATCH) —

    records per sweep

  • on_error (#call, nil) (defaults to: nil) —

    handed anything a sweep raised. Worth passing: a relay whose sweeps are all failing is an outbox filling up, and without this the only symptom is messages that never arrive.

    A one-argument callable is given the exception, as it always was. One that also declares exchange: and routing_key: is given where the record was going as well:

    on_error: lambda { |error, exchange:, routing_key:|
tracing.outbox_publish_failed(exchange: exchange, reason: error.message)
logger.warn("outbox stuck on #{exchange}/#{routing_key}: #{error.message}")
}

    The destination is what the telemetry event wants and what an alert is worth routing by — "the outbox cannot reach +orders-events+" is actionable in a way that "a sweep failed" is not. Both are empty strings when the store itself raised, since no record was in hand.



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 188

def initialize(connection, store, interval: DEFAULT_INTERVAL, batch: DEFAULT_BATCH,
               on_error: nil)
  # A record already holds encoded bytes and rendered headers, so what
  # this needs is the raw publish rather than the one that builds an
  # envelope. Taking either a connection or a transport also lets a test
  # drive a relay with no socket anywhere.
  @transport = connection.respond_to?(:transport) ? connection.transport : connection
  @store = store
  @interval = interval.to_f
  @batch = batch
  @on_error = on_error
  @on_error_wants_destination = destination_wanted?(on_error)
  @lock = Mutex.new
  @wake = ConditionVariable.new
  @stopped = false
end

Instance Attribute Details

#batch ⇒ Object (readonly)

Returns the value of attribute batch.



165
166
167
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 165

def batch
  @batch
end

#interval ⇒ Object (readonly)

Returns the value of attribute interval.



165
166
167
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 165

def interval
  @interval
end

Instance Method Details

#close ⇒ Object

Stops sweeping and waits for the sweep in progress.



227
228
229
230
231
232
233
234
235
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 227

def close
  @lock.synchronize do
    @stopped = true
    @wake.broadcast
  end
  @thread&.join
  @thread = nil
  nil
end

#running? ⇒ Boolean

Whether the sweeping thread is running.

Returns:

  • (Boolean) 


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

def running? = !@thread.nil?

#start ⇒ Object

Sweeps the outbox until #close.



206
207
208
209
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 206

def start
  @thread ||= Thread.new { run }
  self
end

#sweep ⇒ Integer

Publishes one batch, and returns how many went out.

Public so a test can drive a relay without waiting for a tick, and so an application can flush its outbox on demand — at the end of a request, say, rather than up to an interval later.

Returns:

  • (Integer) —

    records published

Raises:

  • (StandardError) —

    whatever the broker or the store raised



219
220
221
 
# File 'lib/acemq/amqp/patterns/outbox.rb', line 219

def sweep
  drain { |error, _record| raise error }
end