AMQP: Using the alternative approach for publisher confirms and returns with an integration flow that serves as gateway

[rainerfrey-inxmail]

When using AMQP outbound adapters, there is an alternative approach to publisher confirms and returns, that involves adding correlation data to the message headers, and wait for the future of this correlation data to fulfil. The application code using this approach deals with messages and headers. In our applications, we typically use Gateway-Interfaces that are bound to integration flows, so that the application doesn't deal with messaging specific APIs in the business logic. Is there any way to implement this approach within the integration flow, eg. with an integration component that acts as a kind of interceptor that handles the correlation data and the future? Does anyone have an example or a rough sketch how that could look like?

[artembilan]

Not fully clear what is your question.
There logic there in the AmqpOutboundEndpoint is like this:

			send(exchangeName, routingKey, requestMessage, correlationData);
			if (this.waitForConfirm && correlationData != null) {
				waitForConfirm(requestMessage, correlationData);
			}

where that correlationData is built from the message ID, AmqpHeaders.PUBLISH_CONFIRM_CORRELATION header or any custom data from the message based on the expression:

	/**
	 * Set a SpEL expression to evaluate confirm correlation at runtime.
	 * @param confirmCorrelationExpression the expression to use.
	 * @since 4.3
	 */
	public void setConfirmCorrelationExpression(Expression confirmCorrelationExpression) {

So, with the @MessagingGateway, on its @Gateway method you are free to put any extra argument marked with some @Header to reflect AmqpOutboundEndpoint expectations.

What kind of interceptor you'd like to see if such a logic is handled internally in the AmqpOutboundEndpoint?

[rainerfrey-inxmail]

I'll try to find a better way to express the question. If I provide an extra argument in the gateway method with @Header("AmqpHeaders.PUBLISH_CONFIRM_CORRELATION"), the caller of this method

• creates a correlation data
• calls the method with it as the extra argument
• waits for the correlation data's future
• if the future times out OR the correlation data contains a returned message, handle it as a failure

As this deals with a messaging, and AMQP specific aspect, I would like if there was a way to implement the above logic in some (loosely termed) interceptor within the flow instead of leaving it to the caller of the gateway method.

[rainerfrey-inxmail]

But it is quite possible that I'm misunderstanding or overlooking something as to why this is an absurd idea.

[artembilan]

Sure! You can avoid that gateway's argument and have a header-enricher just after to put some custom data being represented as a correlation.
Such an info still can be based on the input from the gateway's method.