Class EsbOutboxRelay

java.lang.Object
com.kingsrook.qqq.esb.outbox.EsbOutboxRelay
public final class EsbOutboxRelay extends Object
Publishes the ESB outbox's PENDING rows, and purges old SENT outbox rows and old processed-message rows. Runs on every node, on one daemon thread (so it never blocks JVM exit); EsbRuntimeService starts and stops it. Each pass, for each destination with PENDING rows: - Claim: in one short transaction, lock the destination's first batchSize PENDING rows (SELECT ... FOR UPDATE SKIP LOCKED, where the backend has it - see SkipLockedClaims) and lease them to this node (leaseOwner, leaseExpiresAt). Only the destination's head-of-line rows are claimed: if the oldest PENDING row is locked or leased by another node, or waiting out a retry backoff, the destination is skipped this pass - so each destination's rows are sent in order (by id), by one node at a time. - Send the claimed rows in one broker transaction, each with its row's uuid as its qqq_message_id, and mark them SENT. - On failure, release the lease. If the provider is connected, the failure counts as an attempt on the head row (which is then retried alone, after a doubling backoff, and goes DEAD after maxAttempts); if the broker is unreachable it doesn't count, and the rows wait for it. Delivery is at-least-once: a send whose commit reached the broker, but whose rows weren't marked SENT (crash, lost DB connection, expired lease), is sent again with the same message id - consumers dedupe on it. Backends without SKIP LOCKED rely on the leases alone, which is safe for one relay at a time; run more than one node only on a backend that has it (PostgreSQL).

  • Method Summary

    Modifier and Type
    Method
    Description
    static EsbOutboxRelay
    getInstance()
    Get the singleton instance.
    String
    getNodeId()
    This node's lease owner id.
    boolean
    isRunning()
    Whether the relay thread is running.
    void
    nudge()
    Wake the relay now (e.g., new outbox rows were committed on this node), rather than at its next poll.
    int
    purgeOnce()
    Purge SENT outbox rows, and processed-message rows, older than their retention (each, if enabled).
    int
    relayOnce()
    One pass over every destination with PENDING rows (in the QContext's instance).
    void
    start(QInstance qInstance)
    Start the relay thread, if the instance has the outbox or dedupe enabled (and it isn't already running).
    void
    stop()
    Stop the relay thread, waiting (up to 5 s) for its current pass to end.

    Methods inherited from class java.lang.Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Method Details

    • getInstance

      public static EsbOutboxRelay getInstance()
      Get the singleton instance.

    • start

      public void start(QInstance qInstance)
      Start the relay thread, if the instance has the outbox or dedupe enabled (and it isn't already running). Doesn't block.

    • stop

      public void stop()
      Stop the relay thread, waiting (up to 5 s) for its current pass to end. Rows it had claimed but not sent are released when their lease expires. No-op if not running.

    • isRunning

      public boolean isRunning()
      Whether the relay thread is running.

    • nudge

      public void nudge()
      Wake the relay now (e.g., new outbox rows were committed on this node), rather than at its next poll.

    • getNodeId

      public String getNodeId()
      This node's lease owner id.

    • relayOnce

      public int relayOnce() throws QException
      One pass over every destination with PENDING rows (in the QContext's instance). Returns how many rows were sent. Does nothing (returns 0) if the outbox is not enabled.
      Throws:
      QException

    • purgeOnce

      public int purgeOnce() throws QException
      Purge SENT outbox rows, and processed-message rows, older than their retention (each, if enabled). Returns how many rows were deleted.
      Throws:
      QException