RabbitMQ Generic delayed retries fault handler with message TTL

[scrocquesel]

When an exception occurs, I would like to delayed the retry attempt at rabbitmq side. AFAIK, this can be done in two different ways.

1. Set the DLQ on the main queue to a 'delayed' queue with a fixed TTL, and set the DLQ of the 'delayed' queue to the main queue. If the message is nack, it will go to the delayed queue and will return back to main queue after TTL period. When retry attempts are exhausted, the message can be either discarded with an ack or cloned to a dead letter queue and then ack'ed.
2. The opposite strategy is to clone the message to a delayed queue first and when retries are exhausted, nack the message to send it to DLQ or ack the message to discard it. The pros here is that a per-message TTL can be set when sending the clone message.

In both cases, the message should be clone and sent to an exchange.

I'm trying to implement such a logic that is more or less the KafkaDeadLetterQueue logic in a PublisherDecorator but I'm facing issues.

I cannot use a generic Emitter<?> (do not compile) and if I try to use Emitter<Object> it fails to bind the channel at startup. From the Message object, I cannot access the name of the queue the message is read from to use it as a routing key. And I can't found out how to chain the acknowledgement between the new emitted message and the one that should be discarded (I'm really new to reacive stuff).

Do you think it is something possible to do with a PublisherDecorator and Emitter ? Or a pattern enough generic to be implemented in the connector ?

This is what I have right now :

@ApplicationScoped
public class DelayedRetryStrategyPublisherDecorator implements PublisherDecorator {

    @Channel("delayed")
    private Emitter<Object> emitter;

    @Override
    public int getPriority() {
        return DEFAULT_PRIORITY - 1;
    }

    private static final String PER_MESSSAGE_TTL = "60000";

    @Override
    public Multi<? extends Message<?>> decorate(Multi<? extends Message<?>> publisher, String channelName,
            boolean isConnector) {

        Multi<? extends Message<?>> multi = publisher;
        if (isConnector) {
            multi = multi.map(m -> {
                Optional<IncomingRabbitMQMetadata> maybeRabbitmqMetadata = m
                        .getMetadata(IncomingRabbitMQMetadata.class);
                if (maybeRabbitmqMetadata.isEmpty())
                    return m;

                Message<?> decoratedMessage = m.withNack(t -> {
                    // handle retries if attempt left
                    if (hasAttemptLeft(m)) {

                        var delayedMetadata = OutgoingRabbitMQMetadata.builder()
                                .withExpiration(PER_MESSSAGE_TTL)
                                // .withRoutingKey("") // TODO
                                // TODO decrement attempt counter in header
                                .build();
                        Message<?> delayedMessage = Message.of(m.getPayload(), Metadata.of(delayedMetadata));

                        emitter.send(delayedMessage);

                        // TODO when delayedMessage is received at broker side
                        // discard the message
                        return m.ack();
                    } else {
                        // attempts exhausted, really nack
                        return m.nack(t);
                    }
                });

                return decoratedMessage;
            });
        }

        return multi;
    }

    private boolean hasAttemptLeft(Message<?> m) {
        return true;
    }
}