Package org.acemq.amqp.patterns

Class JdbcOutboxStore

java.lang.Object
org.acemq.amqp.patterns.JdbcOutboxStore
All Implemented Interfaces:
OutboxStore
public final class JdbcOutboxStore extends Object implements OutboxStore
An outbox kept in a relational table, in the same database as the work it accompanies.

Two sources of connections, because the two halves of this class run in different worlds. Writing a record belongs to the caller's transaction and borrows its connection through a ConnectionSupplier; the connection is used and left open, since the transaction that owns it is still running. Everything the relay does — claiming, marking, counting — happens on a background thread with no ambient transaction, so it takes its own connection from a DataSource and closes it. Collapsing the two into one source is the mistake this separation exists to make impossible: a store that inserted on its own auto-committing connection would give up the atomicity that is the entire point.

Claiming is done with a lease rather than SELECT FOR UPDATE. A held lock lasts as long as its transaction, which means a relay that dies mid-batch either strands its rows until the database notices the connection has gone or holds a transaction open across a network publish, and neither is acceptable. A lease is a timestamp: it expires on its own, no matter how the holder died, and the rows return to circulation without anyone intervening.

Claiming happens in two statements, and the second is what makes it safe. Candidate rows are selected, then each is updated with the lease still conditional on being unclaimed. Two relays that select the same candidates will both try to update them; row locking serialises the updates, the loser re-evaluates its condition against the committed row, and matches nothing. The claim is therefore decided by the update's row count, never by the select.

  • Constructor Details

    • JdbcOutboxStore

      public JdbcOutboxStore(ConnectionSupplier transactionalConnection, DataSource relayConnections)
      Parameters:
      transactionalConnection - where add(org.acemq.amqp.api.OutboxRecord) gets the caller's transaction from; see ConnectionSupplier for what this must and must not be
      relayConnections - connections for the relay's own work, taken and closed per call

    • JdbcOutboxStore

      public JdbcOutboxStore(ConnectionSupplier transactionalConnection, DataSource relayConnections, String table, int maxAttempts)
      Parameters:
      transactionalConnection - where add(org.acemq.amqp.api.OutboxRecord) gets the caller's transaction from
      relayConnections - connections for the relay's own work
      table - table holding the outbox
      maxAttempts - how many failures a record may accumulate before it stops being claimed and stays for someone to look at

  • Method Details

    • createSchemaIfAbsent

      public void createSchemaIfAbsent()
      Creates the outbox table if it is not already there.

      For development and tests. In production the table belongs in the migration tool that owns the rest of the schema: a library that creates tables at start-up has taken a decision about when your database changes that is not its to take.

      Throws:
      AceMqException - if the schema cannot be created

    • add

      public void add(OutboxRecord record)
      Description copied from interface: OutboxStore
      Writes a message into the outbox, inside the caller's transaction.

      Must participate in whatever transaction the calling code has open, so that the message is committed or rolled back with the work that produced it. This is the whole pattern; an implementation that cannot honour it should throw rather than pretend.

      Specified by:
      add in interface OutboxStore
      Parameters:
      record - the message to publish once the surrounding work commits

    • claimBatch

      public List<OutboxRecord> claimBatch(int batchSize, Duration lease)
      Description copied from interface: OutboxStore
      Takes a batch of unpublished messages for this relay to work on.

      Claiming is by lease rather than by lock, so that a relay which dies holding a batch does not strand it: the lease expires and another relay picks the messages up. The lease must therefore be comfortably longer than a batch takes to publish, and shorter than the delay anyone would accept before a stalled message moves.

      Specified by:
      claimBatch in interface OutboxStore
      Parameters:
      batchSize - most records to claim
      lease - how long this claim holds before other relays may take the records
      Returns:
      claimed records in the order they were created, oldest first; empty when there is nothing to do

    • markPublished

      public void markPublished(String id)
      Description copied from interface: OutboxStore
      Records that a claimed message reached the broker and needs no further attention.
      Specified by:
      markPublished in interface OutboxStore
      Parameters:
      id - identifier of the published record

    • markFailed

      public void markFailed(String id, String reason)
      Description copied from interface: OutboxStore
      Records that an attempt failed, so the message is tried again later.

      Increments the attempt count and releases the claim. A record that has failed more than the store's limit stops being claimed and stays for someone to look at, which is the right outcome: a message that cannot be published after many tries is a problem to be seen, not one to be retried into the next outage.

      Specified by:
      markFailed in interface OutboxStore
      Parameters:
      id - identifier of the record that failed
      reason - why it failed, for whoever reads the table later

    • pendingCount

      public long pendingCount()
      Specified by:
      pendingCount in interface OutboxStore
      Returns:
      how many messages are waiting to be published, including those currently claimed and those that have exhausted their attempts

    • purgePublished

      public int purgePublished(Duration olderThan)
      Removes published records older than the given age.

      Nothing deletes them otherwise, and an outbox table that only grows eventually makes the relay's own query slow. How long to keep them is a judgement about auditing rather than about messaging, so it is the caller's to make.

      Parameters:
      olderThan - how long a published record is kept
      Returns:
      how many records were removed

    • toString

      public String toString()
      Overrides:
      toString in class Object