M
- An implementation of Message
contained in the dead-letters
within this queue.public class JpaSequencedDeadLetterQueue<M extends EventMessage<?>> extends Object implements SequencedDeadLetterQueue<M>
SequencedDeadLetterQueue
, used for storing dead letters containing
Eventmessages
durably as a DeadLetterEntry
.
Keeps the insertion order intact by saving an incremented index within each unique sequence, backed by the
DeadLetterEntry.getSequenceIndex()
property. Each sequence is uniquely identified by the sequence identifier, stored
in the DeadLetterEntry.getSequenceIdentifier()
field.
When processing an item, single execution across all applications is guaranteed by setting the
DeadLetterEntry.getProcessingStarted()
property, locking other processes out of the sequence for the
configured claimDuration
(30 seconds by default).
The stored entries
are converted to a JpaDeadLetter
when they need to be processed or
filtered. In order to restore the original EventMessage
a matching DeadLetterJpaConverter
is used.
The default supports all EventMessage
implementations provided by the framework. If you have a custom
variant, you have to build your own.
upcasters
are not supported by this implementation, so
breaking changes for events messages stored in the queue should be avoided.
Modifier and Type | Class and Description |
---|---|
static class |
JpaSequencedDeadLetterQueue.Builder<T extends EventMessage<?>>
Builder class to instantiate an
JpaSequencedDeadLetterQueue . |
Modifier | Constructor and Description |
---|---|
protected |
JpaSequencedDeadLetterQueue(JpaSequencedDeadLetterQueue.Builder<T> builder)
Instantiate a JPA
SequencedDeadLetterQueue based on the given builder . |
Modifier and Type | Method and Description |
---|---|
long |
amountOfSequences()
Returns the number of unique sequences contained in this queue.
|
static <M extends EventMessage<?>> |
builder()
Creates a new builder, capable of building a
JpaSequencedDeadLetterQueue according to the provided
configuration. |
void |
clear()
Clears out all
dead letters present in this queue. |
boolean |
contains(Object sequenceIdentifier)
Check whether there's a sequence of
dead letters for the given sequenceIdentifier . |
Iterable<Iterable<DeadLetter<? extends M>>> |
deadLetters()
Return all
dead letter sequences held by this queue. |
Iterable<DeadLetter<? extends M>> |
deadLetterSequence(Object sequenceIdentifier)
Return all the
dead letters for the given sequenceIdentifier in insert order. |
void |
enqueue(Object sequenceIdentifier,
DeadLetter<? extends M> letter)
Enqueues a
dead letter containing an implementation of M to this queue. |
void |
evict(DeadLetter<? extends M> letter)
Evict the given
letter from this queue. |
boolean |
isFull(Object sequenceIdentifier)
Validates whether this queue is full for the given
sequenceIdentifier . |
boolean |
process(Function<DeadLetter<? extends M>,EnqueueDecision<M>> processingTask)
Process a sequence of enqueued
dead letters with the given processingTask . |
boolean |
process(Predicate<DeadLetter<? extends M>> sequenceFilter,
Function<DeadLetter<? extends M>,EnqueueDecision<M>> processingTask)
Process a sequence of enqueued
dead letters through the given processingTask matching
the sequenceFilter . |
void |
requeue(DeadLetter<? extends M> letter,
UnaryOperator<DeadLetter<? extends M>> letterUpdater)
Reenters the given
letter , updating the contents with the letterUpdater . |
long |
sequenceSize(Object sequenceIdentifier)
Returns the number of dead letters for the sequence matching the given
sequenceIdentifier contained in
this queue. |
long |
size()
Returns the number of dead letters contained in this queue.
|
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
enqueueIfPresent
protected JpaSequencedDeadLetterQueue(JpaSequencedDeadLetterQueue.Builder<T> builder)
SequencedDeadLetterQueue
based on the given builder
.builder
- The JpaSequencedDeadLetterQueue.Builder
used to instantiate a JpaSequencedDeadLetterQueue
instance.public static <M extends EventMessage<?>> JpaSequencedDeadLetterQueue.Builder<M> builder()
JpaSequencedDeadLetterQueue
according to the provided
configuration. Note that the JpaSequencedDeadLetterQueue.Builder.processingGroup(String)
, JpaSequencedDeadLetterQueue.Builder.transactionManager
,
JpaSequencedDeadLetterQueue.Builder.serializer(Serializer)
and JpaSequencedDeadLetterQueue.Builder.entityManagerProvider
are mandatory for the queue
to be constructed.M
- An implementation of Message
contained in the dead-letters
within this queue.public void enqueue(@Nonnull Object sequenceIdentifier, @Nonnull DeadLetter<? extends M> letter) throws DeadLetterQueueOverflowException
SequencedDeadLetterQueue
dead letter
containing an implementation of M
to this queue.
The dead letter
will be appended to a sequence depending on the sequenceIdentifier
. If there is
no sequence yet, it will construct one.
enqueue
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
sequenceIdentifier
- The identifier of the sequence the letter
belongs to.letter
- The DeadLetter
to enqueue.DeadLetterQueueOverflowException
- Thrown when this queue is full
.public void evict(DeadLetter<? extends M> letter)
SequencedDeadLetterQueue
letter
from this queue. Nothing happens if the dead letter
does not
exist in this queue.evict
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
letter
- The dead letter
to evict from this queue.public void requeue(@Nonnull DeadLetter<? extends M> letter, @Nonnull UnaryOperator<DeadLetter<? extends M>> letterUpdater) throws NoSuchDeadLetterException
SequencedDeadLetterQueue
letter
, updating the contents with the letterUpdater
. This method should be
invoked if processing
decided to keep the letter in the queue.
This operation adjusts the DeadLetter.lastTouched()
. It may adjust the DeadLetter.cause()
and
DeadLetter.diagnostics()
, depending on the given letterUpdater
.
requeue
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
letter
- The dead letter
to reenter in this queue.letterUpdater
- A lambda
taking in the given letter
and updating the entry for
requeueing. This may adjust the DeadLetter.cause()
and
DeadLetter.diagnostics()
, for example.NoSuchDeadLetterException
- Thrown if the given letter
does not exist in the queue.public boolean contains(@Nonnull Object sequenceIdentifier)
SequencedDeadLetterQueue
dead letters
for the given sequenceIdentifier
.contains
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
sequenceIdentifier
- The identifier used to validate for contained dead letters
instances.true
if there are dead letters
present for the given
sequenceIdentifier
, false
otherwise.public Iterable<DeadLetter<? extends M>> deadLetterSequence(@Nonnull Object sequenceIdentifier)
SequencedDeadLetterQueue
dead letters
for the given sequenceIdentifier
in insert order.deadLetterSequence
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
sequenceIdentifier
- The identifier of the sequence of dead letters
to return.dead letters
for the given sequenceIdentifier
in insert order.public Iterable<Iterable<DeadLetter<? extends M>>> deadLetters()
SequencedDeadLetterQueue
dead letter
sequences held by this queue. The sequences are not necessarily
returned in insert order.deadLetters
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
dead letter
sequences held by this queue.public boolean isFull(@Nonnull Object sequenceIdentifier)
SequencedDeadLetterQueue
sequenceIdentifier
.
This method returns true
either when the maximum amount of sequences or the maximum sequence size is
reached.
isFull
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
sequenceIdentifier
- The identifier of the sequence to validate for.true
either when the limit of this queue is reached. Returns false
otherwise.public boolean process(@Nonnull Predicate<DeadLetter<? extends M>> sequenceFilter, @Nonnull Function<DeadLetter<? extends M>,EnqueueDecision<M>> processingTask)
SequencedDeadLetterQueue
dead letters
through the given processingTask
matching
the sequenceFilter
. Will pick the oldest available sequence based on the DeadLetter.lastTouched()
field from every sequence's first entry.
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 EnqueueDecision
returned by the processingTask
to decide whether to
SequencedDeadLetterQueue.evict(DeadLetter)
or SequencedDeadLetterQueue.requeue(DeadLetter, UnaryOperator)
a dead letter from the selected
sequence. The processingTask
is invoked as long as letters are present in the selected sequence and the
result of processing returns false
for EnqueueDecision.shouldEnqueue()
decision. The latter means
the dead letter should be evicted.
This operation protects against concurrent invocations of the processingTask
on the filtered sequence.
Doing so ensure enqueued messages are handled in order.
process
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
sequenceFilter
- A lambda
selecting the sequences within this queue to process with the
processingTask
.processingTask
- A function processing a dead letter
. Returns a EnqueueDecision
used to deduce whether to SequencedDeadLetterQueue.evict(DeadLetter)
or
SequencedDeadLetterQueue.requeue(DeadLetter, UnaryOperator)
the dead letter.true
if an entire sequence of dead letters
was processed successfully,
false
otherwise. This means the processingTask
processed all dead letters
of a
sequence and the outcome was to evict each instance.public boolean process(@Nonnull Function<DeadLetter<? extends M>,EnqueueDecision<M>> processingTask)
SequencedDeadLetterQueue
dead letters
with the given processingTask
. Will pick
the oldest available sequence based on the DeadLetter.lastTouched()
field from every sequence's first
entry.
Note that only a single matching sequence is processed!
Uses the EnqueueDecision
returned by the processingTask
to decide whether to
SequencedDeadLetterQueue.evict(DeadLetter)
or SequencedDeadLetterQueue.requeue(DeadLetter, UnaryOperator)
the dead letter. The
processingTask
is invoked as long as letters are present in the selected sequence and the result of
processing returns false
for EnqueueDecision.shouldEnqueue()
decision. The latter means the dead
letter should be evicted.
This operation protects against concurrent invocations of the processingTask
on the filtered sequence. *
Doing so ensure enqueued messages are handled in order.
process
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
processingTask
- A function processing a dead letter
. Returns a EnqueueDecision
used to deduce whether to SequencedDeadLetterQueue.evict(DeadLetter)
or
SequencedDeadLetterQueue.requeue(DeadLetter, UnaryOperator)
the dead letter.true
if an entire sequence of dead letters
was processed successfully,
false
otherwise. This means the processingTask
processed all dead letters
of a
sequence and the outcome was to evict each instance.public void clear()
SequencedDeadLetterQueue
dead letters
present in this queue.clear
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
public long sequenceSize(@Nonnull Object sequenceIdentifier)
SequencedDeadLetterQueue
sequenceIdentifier
contained in
this queue.
Note that there's a window of opportunity where the size might exceed the maximum sequence size to accompany concurrent usage.
sequenceSize
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
sequenceIdentifier
- The identifier of the sequence to retrieve the size from.sequenceIdentifier
.public long size()
SequencedDeadLetterQueue
size
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
public long amountOfSequences()
SequencedDeadLetterQueue
Note that there's a window of opportunity where the size might exceed the maximum amount of sequences to accompany concurrent usage of this dead letter queue.
amountOfSequences
in interface SequencedDeadLetterQueue<M extends EventMessage<?>>
Copyright © 2010–2023. All rights reserved.