Package org.acemq.amqp.core

Class ConsumerOptions

java.lang.Object
org.acemq.amqp.core.ConsumerOptions
public final class ConsumerOptions extends Object
How a consumer should behave.

  • Method Details

    • defaults

      public static ConsumerOptions defaults()
      Returns:
      a prefetch of 50 with failures rejected rather than requeued
      Implementation Note:
      failures are not requeued by default. Requeueing a message that just failed sends it straight back to the same consumer, which spins at full speed and starves every other message. Until the retry ladder exists, rejecting is the honest behaviour: the message goes to a dead-letter queue if one is configured.

    • prefetch

      public static ConsumerOptions prefetch(int prefetch)
      Parameters:
      prefetch - maximum unacknowledged deliveries at once
      Returns:
      options with that prefetch

    • requeueOnFailure

      public ConsumerOptions requeueOnFailure()
      Requeues a message when its handler fails.

      Only safe when the handler is expected to succeed on another consumer, such as when shedding load. Using it for ordinary failures produces a hot loop.

      Returns:
      options that requeue on failure

    • withRetry

      public ConsumerOptions withRetry(RetryPolicy retryPolicy)
      Retries failures on a schedule, never in a sleeping handler.

      Turning this on makes the consumer declare a dead-letter queue, a parking lot, and one retry rung per distinct delay the policy wants the broker to hold — see RetryPolicy.brokerWaitThreshold(). A schedule that runs in a few seconds needs no rungs at all, and gets none.

      A failed message is republished either way, with its attempt counter advanced. Above the threshold it goes into the rung and comes back when the time-to-live expires; below it the engine holds the delivery for the wait and puts it back on the source queue. Once the attempts or the age limit are used up it lands in the dead-letter queue with the reason attached.

      Parameters:
      retryPolicy - the schedule to follow
      Returns:
      options with retries enabled

    • idempotent

      public ConsumerOptions idempotent(IdempotencyStore store)
      Handles each message at most once, by remembering which have been handled.

      Brokers deliver at least once, so a duplicate is a normal event rather than a fault: a consumer that dies between doing the work and acknowledging it will see the message again. With a store configured, the second delivery is acknowledged without running the handler.

      This makes the delivery idempotent, not the handler. Work the handler does outside the store's knowledge — a row written, a payment taken — is only protected to the extent that the store and that work fail together, which is why a store sharing the handler's database is stronger than one in memory.

      Parameters:
      store - where handled identifiers are remembered
      Returns:
      options with deduplication enabled

    • as

      public ConsumerOptions as(Codec codec)
      Reads this queue with one named format instead of whatever arrives.

      Almost never needed. A consumer normally says nothing about format, because the content type on each message picks the codec, and that is what lets a producer change format without a consumer change.

      It is needed for Avro and Protobuf, and the reason is the same one that stops them having an asAvro(): their bytes carry no description of themselves, so a reader that has not been told the schema cannot construct one on seeing the message. Self describing formats need nothing here; schema-bound formats must be named at both ends.

      Parameters:
      codec - the format to read with
      Returns:
      a copy of these options reading that format

    • codec

      public Optional<Codec> codec()
      Returns:
      the format this consumer was told to read, if it was told one

    • idempotencyStore

      public Optional<IdempotencyStore> idempotencyStore()
      Returns:
      the idempotency store, when one is configured

    • retryPolicy

      public Optional<RetryPolicy> retryPolicy()
      Returns:
      the retry schedule, when one is configured

    • prefetch

      public int prefetch()

    • isRequeueOnFailure

      public boolean isRequeueOnFailure()