Interface SequencedDeadLetterQueue<M extends Message>
- Type Parameters:
M- An implementation ofMessagecontained in thedead letterswithin this queue.
- All Known Implementing Classes:
CachingSequencedDeadLetterQueue,InMemorySequencedDeadLetterQueue,JdbcSequencedDeadLetterQueue,JpaSequencedDeadLetterQueue
The contained sequences are uniquely identifiable through the "sequence identifier." Dead-letters are kept in the
form of a DeadLetter. It is highly recommended to use the
process operation (or any of its variants) to consume
letters from the queue for retrying. This method ensures sequences cannot be concurrently accessed, thus protecting
the user against handling messages out of order.
All methods in this interface return CompletableFuture to support asynchronous implementations. In-memory
implementations may simply return completed futures, while persistent implementations (JPA, JDBC, etc.) can leverage
the async nature for non-blocking I/O.
- Since:
- 4.6.0
- Author:
- Steven van Beelen, Allard Buijze, Milan Savic, Mitchell Herrijgers
- See Also:
-
Method Summary
Modifier and TypeMethodDescriptionamountOfSequences(@Nullable ProcessingContext context) Returns the number of unique sequences contained in this queue.clear(@Nullable ProcessingContext context) Clears out alldead letterspresent in this queue.contains(Object sequenceIdentifier, @Nullable ProcessingContext context) Check whether there's a sequence ofdead lettersfor the givensequenceIdentifier.CompletableFuture<Iterable<Iterable<DeadLetter<? extends M>>>> deadLetters(@Nullable ProcessingContext context) Return alldead lettersequences held by this queue.CompletableFuture<Iterable<DeadLetter<? extends M>>> deadLetterSequence(Object sequenceIdentifier, @Nullable ProcessingContext context) Return all thedead lettersfor the givensequenceIdentifierin insert order.enqueue(Object sequenceIdentifier, DeadLetter<? extends M> letter, @Nullable ProcessingContext context) Enqueues adead lettercontaining an implementation ofMto this queue.default CompletableFuture<Boolean> enqueueIfPresent(Object sequenceIdentifier, Supplier<DeadLetter<? extends M>> letterBuilder, @Nullable ProcessingContext context) Enqueue the result of the givenletterBuilderonly if there already are otherdead letterswith the samesequenceIdentifierpresent in this queue.evict(DeadLetter<? extends M> letter, @Nullable ProcessingContext context) Evict the givenletterfrom this queue.isFull(Object sequenceIdentifier, @Nullable ProcessingContext context) Validates whether this queue is full for the givensequenceIdentifier.default CompletableFuture<Boolean> process(Function<DeadLetter<? extends M>, CompletableFuture<EnqueueDecision<M>>> processingTask, @Nullable ProcessingContext context) Process a single sequence of enqueueddead letterswith the givenprocessingTask.process(Predicate<DeadLetter<? extends M>> sequenceFilter, Function<DeadLetter<? extends M>, CompletableFuture<EnqueueDecision<M>>> processingTask, @Nullable ProcessingContext context) Process a single sequence of enqueueddead lettersthrough the givenprocessingTaskmatching thesequenceFilter.requeue(DeadLetter<? extends M> letter, UnaryOperator<DeadLetter<? extends M>> letterUpdater, @Nullable ProcessingContext context) Reenters the givenletter, updating the contents with theletterUpdater.sequenceSize(Object sequenceIdentifier, @Nullable ProcessingContext context) Returns the number of dead letters for the sequence matching the givensequenceIdentifiercontained in this queue.size(@Nullable ProcessingContext context) Returns the number of dead letters contained in this queue.
-
Method Details
-
enqueue
CompletableFuture<Void> enqueue(Object sequenceIdentifier, DeadLetter<? extends M> letter, @Nullable ProcessingContext context) Enqueues adead lettercontaining an implementation ofMto this queue.The
dead letterwill be appended to a sequence depending on thesequenceIdentifier. If there is no sequence yet, it will construct one.- Parameters:
sequenceIdentifier- The identifier of the sequence theletterbelongs to.letter- TheDeadLetterto enqueue.context- the processing context within which to enqueue the givenletter- Returns:
- A
CompletableFuturethat completes when the operation is done. The future completes exceptionally with aDeadLetterQueueOverflowExceptionwhen this queueis full.
-
enqueueIfPresent
default CompletableFuture<Boolean> enqueueIfPresent(Object sequenceIdentifier, Supplier<DeadLetter<? extends M>> letterBuilder, @Nullable ProcessingContext context) Enqueue the result of the givenletterBuilderonly if there already are otherdead letterswith the samesequenceIdentifierpresent in this queue.- Parameters:
sequenceIdentifier- The identifier of the sequence to store the result of theletterBuilderin.letterBuilder- TheDeadLetterbuilder constructing the letter to enqueue. Only invoked if the givensequenceIdentifieris contained.context- the processing context within which to enqueue the result of the givenletterBuilderif there is a sequence for the givensequenceIdentifier- Returns:
- A
CompletableFuturewithtrueif there aredead lettersfor the givensequenceIdentifierand thus theletterBuilder'soutcome is inserted. Otherwise, the future completes withfalse. The future completes exceptionally with aDeadLetterQueueOverflowExceptionwhen this queue isisFull(Object, ProcessingContext)for the givensequenceIdentifier.
-
evict
Evict the givenletterfrom this queue. Nothing happens if thedead letterdoes not exist in this queue.- Parameters:
letter- Thedead letterto evict from this queue.context- the processing context within which to evict the givenletter- Returns:
- A
CompletableFuturethat completes when the eviction is done.
-
requeue
CompletableFuture<Void> requeue(DeadLetter<? extends M> letter, UnaryOperator<DeadLetter<? extends M>> letterUpdater, @Nullable ProcessingContext context) Reenters the givenletter, updating the contents with theletterUpdater. This method should be invoked ifprocessingdecided to keep the letter in the queue.This operation adjusts the
DeadLetter.lastTouched(). It may adjust theDeadLetter.cause()andDeadLetter.diagnostics(), depending on the givenletterUpdater.- Parameters:
letter- Thedead letterto reenter in this queue.letterUpdater- Alambdataking in the givenletterand updating the entry for requeueing. This may adjust theDeadLetter.cause()andDeadLetter.diagnostics(), for example.context- the processing context within which to requeue the givenletter- Returns:
- A
CompletableFuturethat completes when the requeue is done. The future completes exceptionally with aNoSuchDeadLetterExceptionif the givenletterdoes not exist in the queue.
-
contains
Check whether there's a sequence ofdead lettersfor the givensequenceIdentifier.- Parameters:
sequenceIdentifier- The identifier used to validate for containeddead lettersinstances.context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturewithtrueif there aredead letterspresent for the givensequenceIdentifier,falseotherwise.
-
deadLetterSequence
CompletableFuture<Iterable<DeadLetter<? extends M>>> deadLetterSequence(Object sequenceIdentifier, @Nullable ProcessingContext context) Return all thedead lettersfor the givensequenceIdentifierin insert order.- Parameters:
sequenceIdentifier- The identifier of the sequence ofdead lettersto return.context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturewith all thedead lettersfor the givensequenceIdentifierin insert order.
-
deadLetters
CompletableFuture<Iterable<Iterable<DeadLetter<? extends M>>>> deadLetters(@Nullable ProcessingContext context) Return alldead lettersequences held by this queue. The sequences are not necessarily returned in insert order.- Parameters:
context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturewith alldead lettersequences held by this queue.
-
isFull
Validates whether this queue is full for the givensequenceIdentifier.This method returns
trueeither when the maximum amount of sequences or the maximum sequence size is reached.- Parameters:
sequenceIdentifier- The identifier of the sequence to validate for.context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturewithtrueeither when the limit of this queue is reached. Containsfalseotherwise.
-
size
Returns the number of dead letters contained in this queue.- Parameters:
context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturewith the number of dead letters contained in this queue.
-
sequenceSize
CompletableFuture<Long> sequenceSize(Object sequenceIdentifier, @Nullable ProcessingContext context) Returns the number of dead letters for the sequence matching the givensequenceIdentifiercontained in this queue.Note that there's a window of opportunity where the size might exceed the maximum sequence size to account for concurrent usage.
- Parameters:
sequenceIdentifier- The identifier of the sequence to retrieve the size from.context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturewith the number of dead letters for the sequence matching the givensequenceIdentifier.
-
amountOfSequences
Returns the number of unique sequences contained in this queue.Note that there's a window of opportunity where the size might exceed the maximum amount of sequences to account for concurrent usage of this dead letter queue.
- Parameters:
context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturewith the number of unique sequences contained in this queue.
-
process
CompletableFuture<Boolean> process(Predicate<DeadLetter<? extends M>> sequenceFilter, Function<DeadLetter<? extends M>, CompletableFuture<EnqueueDecision<M>>> processingTask, @Nullable ProcessingContext context) Process a single sequence of enqueueddead lettersthrough the givenprocessingTaskmatching thesequenceFilter. It will pick the oldest available sequence, determined by theDeadLetter.lastTouched()field of the first entry in each sequence.Note that only a single matching sequence is processed! Furthermore, only the first dead letter is validated, because it is the blocker for the processing of the rest of the sequence.
Uses the
EnqueueDecisionreturned by theprocessingTaskto decide whether toevict(DeadLetter, ProcessingContext)orrequeue(DeadLetter, UnaryOperator, ProcessingContext)a dead letter from the selected sequence. TheprocessingTaskis invoked as long as letters are present in the selected sequence and the result of processing returnsfalseforEnqueueDecision.shouldEnqueue()decision. The latter means the dead letter should be evicted.This operation protects against concurrent invocations of the
processingTaskon the filtered sequence. Doing so ensures enqueued messages are handled in order.- Parameters:
sequenceFilter- Alambdaselecting the sequences within this queue to process with theprocessingTask.processingTask- A function processing adead letter. Returns aCompletableFuturewith anEnqueueDecisionused to deduce whether toevict(DeadLetter, ProcessingContext)orrequeue(DeadLetter, UnaryOperator, ProcessingContext)the dead letter.context- theProcessingContextin which the dead letters are processed- Returns:
- A
CompletableFuturewithtrueif an entire sequence ofdead letterswas processed successfully,falseotherwise. This means theprocessingTaskprocessed alldead lettersof a sequence and the outcome was to evict each instance.
-
process
default CompletableFuture<Boolean> process(Function<DeadLetter<? extends M>, CompletableFuture<EnqueueDecision<M>>> processingTask, @Nullable ProcessingContext context) Process a single sequence of enqueueddead letterswith the givenprocessingTask. It will pick the oldest available sequence, determined by theDeadLetter.lastTouched()field of the first entry in each sequence.Note that only a single matching sequence is processed!
Uses the
EnqueueDecisionreturned by theprocessingTaskto decide whether toevict(DeadLetter, ProcessingContext)orrequeue(DeadLetter, UnaryOperator, ProcessingContext)the dead letter. TheprocessingTaskis invoked as long as letters are present in the selected sequence and the result of processing returnsfalseforEnqueueDecision.shouldEnqueue()decision. The latter means the dead letter should be evicted.This operation protects against concurrent invocations of the
processingTaskon the filtered sequence. Doing so ensures enqueued messages are handled in order.- Parameters:
processingTask- A function processing adead letter. Returns aCompletableFuturewith anEnqueueDecisionused to deduce whether toevict(DeadLetter, ProcessingContext)orrequeue(DeadLetter, UnaryOperator, ProcessingContext)the dead letter.context- theProcessingContextin which the dead letters are processed- Returns:
- A
CompletableFuturewithtrueif an entire sequence ofdead letterswas processed successfully,falseotherwise. This means theprocessingTaskprocessed alldead lettersof a sequence and the outcome was to evict each instance.
-
clear
Clears out alldead letterspresent in this queue.- Parameters:
context- theProcessingContextwherein this operation is invoked, if any- Returns:
- A
CompletableFuturethat completes when all dead letters have been cleared.
-