Package org.acemq.amqp.patterns

Class OutboxRelay

java.lang.Object
org.acemq.amqp.patterns.OutboxRelay
All Implemented Interfaces:
AutoCloseable
public final class OutboxRelay extends Object implements AutoCloseable
Moves messages out of the outbox and into the broker.

The half of the pattern that does the sending. It claims a batch, publishes each record and marks it, and it does so on its own thread so that no request ever waits for a broker: the point of writing to a table was to let the request commit and return, and a relay that ran inline would give that back.

The order of the two steps is the only interesting decision, and it is settled by which failure is survivable. Publishing before marking means a relay that dies in between will publish the message again on restart. Marking before publishing means the same crash loses the message entirely. A duplicate can be absorbed by an idempotent consumer; a message that was never sent and is recorded as sent cannot be recovered by anyone. So: publish, then mark, and accept at-least-once as the guarantee on offer.

A failed publish releases the claim and increments the attempt count rather than stopping the batch, because one unroutable message must not hold up the queue behind it. After enough failures the store stops handing the record out and it stays in the table, visible, for someone to look at.

Ordering between messages is preserved only while a single relay runs: records are claimed oldest first and published one at a time. Run two relays for throughput and you have traded ordering for it, which is usually the right trade and never a silent one.

  • Constructor Details

    • OutboxRelay

      public OutboxRelay(AceMq mq, OutboxStore store)
      Parameters:
      mq - connection to publish through
      store - outbox to drain

    • OutboxRelay

      public OutboxRelay(AceMq mq, OutboxStore store, int batchSize, Duration pollInterval, Duration lease)
      Parameters:
      mq - connection to publish through
      store - outbox to drain
      batchSize - how many records to claim at once
      pollInterval - how long to wait between passes when there is nothing to do
      lease - how long a claim holds; must comfortably exceed the time a batch takes to publish, or another relay will start sending the same messages alongside this one

  • Method Details

    • start

      public void start()
      Starts draining the outbox on a background thread.

      Calling this twice does nothing the second time.

    • drain

      public int drain()
      Publishes everything currently waiting, in batches.

      Stops after a fixed number of batches even if more remain, so that a large backlog is worked through steadily instead of monopolising the thread and starving the poll that measures progress.

      Returns:
      how many records were published

    • drainOnce

      public int drainOnce()
      Claims one batch and publishes it.
      Returns:
      how many records were published; zero when the outbox is empty

    • published

      public long published()
      Returns:
      how many records this relay has published

    • failed

      public long failed()
      Returns:
      how many publish attempts this relay has seen fail

    • isRunning

      public boolean isRunning()
      Returns:
      whether the background thread is running

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable

    • toString

      public String toString()
      Overrides:
      toString in class Object