Add support for handling returned (undeliverable) messages in Sender

[a701440]

Looks like functionality of installing a Channel.addReturnListener is missing.
Sender.sendWithPublishConfirms does not handle undeliverable messages.

[acogoluegnes]

This should be doable by adding an trackReturtned option to SendOptions and a returned boolean property to OutboundMessageResult (naming is not definitive).

[vanseverk]

@acogoluegnes can I take this issue onto me?

[acogoluegnes]

@vanseverk Sure, thanks!

[vanseverk]

@acogoluegnes I seem to be running into an issue using the returnListener in the Sender class.

It has a method with the signature handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) but does not seem to return any deliveryTag.

This means the unconfirmed message can not be retrieved, won't be passed along the OutboundMessageResult and won't be deleted from the ConcurrentNavigableMap<Long, OutboundMessage> unconfirmed.

[acogoluegnes]

It is possible to create a OutboundMessage from the values in handleReturn and wrap it in a OutboundMessageResult to emit it downstream.

Unfortunately, this would not help in cleaning the unconfirmed map with messages that could not be routed. I'd suggest adding the delivery tag into a custom reactor_rabbitmq_delivery_tag header to be able to get it back in the return listener. @vanseverk WDYT?

[vanseverk]

@acogoluegnes well, adding the delivery tag in a custom header might be a solution to the problem. Would this not mean that we're sending a duplicate one, however? Assuming the current delivery tag is already in a header.

In case we can add and succesfully retrieve (or signal an error in case it's missing) the header, I believe this would indeed be the solution I was considering myself as well.

[acogoluegnes]

The delivery tag does not travel on the wire, it is just a sequence the client and the server increment independently, so there is no duplication here.

[acogoluegnes]

@vanseverk Maybe you'll want to rebase your work on master because of #99, which provided a fix to remove the confirm listener at the end of the flow. This is useful when channel pooling is in use. Note the return listener should be removed as well.

[vanseverk]

@acogoluegnes thanks for the heads-up!

[vanseverk]

@acogoluegnes it seems by default the BasicProperties object in OutboundMessage is "null" by default, and the headers inside of it are empty as well. Assuming you'd use this header map to put the reactor_rabbitmq_delivery_tag into that is a bit of a problem.

Wouldn't it be better to have an BasicProperties (including the headers map) instantiated by default?

I also noticed this TODO in OutboundMessage
// TODO add a correlation property to use for OutboundMessageResult // (instead of using the whole message)

Maybe his correlation property could be used instead somehow? Although not entirely sure how we could then retrieve it in the returnListener's handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body)

[acogoluegnes]

@vanseverk OutboundMessage#properties is null by default because Channel#basicPublish accepts null properties.

I think we can leave the OutboundMessage instance the same, just add the header when publishing: https://github.com/reactor/reactor-rabbitmq/blob/edc3937a036d88d10871475329e3bd13c4aae0e0/src/main/java/reactor/rabbitmq/Sender.java#L635-L640

This would look like the following in pseudo-code:

if (message.getProperties() == null) {
  ... create properties with delivery tag in `reactor_rabbitmq_delivery_tag` header
  channel.basicPublish(
    message.getExchange(),
    message.getRoutingKey(),
    newProps,
    message.getBody()
  );
} else {
  ... create new properties from messages properties, add delivery tag in `reactor_rabbitmq_delivery_tag` header
  channel.basicPublish(
    message.getExchange(),
    message.getRoutingKey(),
    newProps,
    message.getBody()
  );
}

This way we leave the OutboundMessage intact, the delivery tag trick is just an implementation detail. Note creating a copy of existing properties will be a bit tedious, because the properties will have to be copied one by one.

Note also this should be done in ConfirmSendContext as well (used to retry in case of error). This may be a bit trickier and is just polishing, I can do it once your PR is merged.

Forget about the TODO, it must be an old leftover.

The reactor_rabbitmq_delivery_tag header is a correlation ID as well, it should show up in the properties argument of ReturnListener#handleReturn and be used to clean up the unconfirmed map.

[a701440]

BTW does this mean that undeliverable messages would be causing a memory leak now?
Given that they are not removed from that map? Are they removed from the map at some point?