Class EsbTriggerRunner

java.lang.Object
com.kingsrook.qqq.esb.runtime.EsbTriggerRunner
public class EsbTriggerRunner extends Object
Runs one trigger's consumers on this node (spec section 6). Consumers: the runner starts `concurrency` workers, each a virtual thread with its own transacted JMS session and consumer, receiving with a 1 s timeout. Queue triggers use createConsumer (the workers, and other nodes, compete for messages); topic triggers use createSharedDurableConsumer on the trigger's effective subscription name (so each message is processed once across all nodes, and is kept while they are down). Names come from the two-argument EsbTrigger methods, which don't need a QContext. Each message (or, in BATCH mode, each batch: up to batchSize messages, or what arrives within batchWaitMs of the first) is run through EsbTriggerHandler.runProcess on a new virtual thread, so each run starts with a clean QContext, and can be timed out (the run thread is interrupted, and the attempt counts as failed). Then, in the same transacted session: - success: commit (acknowledge); - failure below maxAttempts (the attempt number being the broker's JMSXDeliveryCount - the highest in a batch): wait the backoff, then roll back, so the broker redelivers to this queue or subscription only (while waiting, the worker is busy); - failure at maxAttempts: send each message to the dead-letter queue (or discard it, per onDeadLetter), with qqqError, qqqFailedTrigger, qqqAttempts and qqqFailedAt properties, then commit; - a message that isn't a CloudEvent: dead-letter it at once (qqqError "unparseable message"), without retries. In a batch, the other messages still run. Dead letters copy the body of a text, bytes, map, or stream message. An object message's body is not deserialized, so its dead letter has no body, and a qqqBodyDropped property of true. A batch fails as a unit, so every message in it is dead-lettered when the batch reaches maxAttempts, each with its own attempt count. A session commits or rolls back everything it received, so when a batch's run fails and is retried, the dead letters of its unparseable messages roll back too: those messages are redelivered as well, and dead-lettered by a later commit. Connecting never blocks the caller: until a worker has a consumer, the trigger is CONNECTING, and the worker retries every second, or at once when EsbConnectionManager reports a reconnect (QEsbRuntime passes that on). When the connection is lost, workers close their consumers and connect again. pauseLocal, resumeLocal and restartLocal act on this node only (see EsbTriggerControl for every node). Pausing closes the consumers - each after the run it is in, if any, and at most one receive timeout (1 s) after the pause - so messages stay on the broker. A batch being collected is cut short by a pause, and runs with what it has. Restarting makes each worker build a new session and consumer. Workers block in the broker client's receive, which (on Java 21) holds its virtual thread's carrier; the JDK adds carrier threads to make up for that, up to its limit (jdk.virtualThreadScheduler.maxPoolSize, 256 by default) - so a node's total concurrency, across all triggers, should stay well below it. Stopping (QEsbRuntime.stop) interrupts runs in progress, and rolls their messages back, so they are redelivered after the next start - not lost, and not counted as failures. A run that ignores the interrupt is waited for up to 2 s; after that its message is rolled back anyway, and the run is left to finish on its own (with at-least-once delivery, the message is then processed again).

  • Method Details

    • getState

      public EsbTriggerState getState()
      This trigger's state on this node. RUNNING needs every worker to have a consumer built since the last restart or reconnect - so right after restartLocal, the state is CONNECTING until the consumers are rebuilt.

    • pauseLocal

      public void pauseLocal()
      Pause this trigger on this node: its consumers close (see the class comment), and messages wait on the broker. No-op if stopped or paused.

    • resumeLocal

      public void resumeLocal()
      Resume this trigger on this node. No-op if stopped or not paused.

    • restartLocal

      public void restartLocal()
      Restart this trigger's consumers on this node: each worker closes its session and consumer (after its current run, if any) and opens new ones. A paused trigger stays paused. No-op if stopped.

    • getTriggerName

      public String getTriggerName()
      The trigger's name: processName.destinationName.

    • getProcessName

      public String getProcessName()
      The name of the process the trigger runs.

    • getTrigger

      public EsbTrigger getTrigger()
      The trigger's meta-data.

    • getDestination

      public QEsbDestinationMetaData getDestination()
      The trigger's destination (queue or topic).