JmsOutboundGateway as ServiceActivator - for async gateway AdviceChains doesn't work

[LukaszPa]

Hi,

   I'm using JmsOutboundGateway as  ServiceActivator  to input and then collect messages from queue. 

If i use JmsOutboundGateway with setAsync(true) then RetryAdvice inputted in gateway AdviceChain doesn't work.

  When I set  setAsync(false) in JmsOutboundGateway,  RetryAdvice  inputted in gateway AdviceChain works fine -  but i will stuck on sync flow.  :(

 Is it possible to use  RetryAdvice  added to gateway AdviceChain  in Async mode?   

Best Regards,
Luke

[artembilan]

No, it is not. That advice is applied around handleRequestMessage() which in case of async mode returns a CompletableFuture immediately and all the "hard" JMS logic happens on a different thread.
We cannot retry an async mode until we come up with an async retry feature, which possible could be as a composition on that returned CompletableFuture: spring-projects/spring-retry#81.

We may borrow some ideas from AWS SDK v2 AsyncRetryableStage...

But for now, no, there is no way to retry async mode.

We may consider, though, to use an async mode combination with a FluxMessageChannel as an output and then reactive() for downstream handler:

	/**
	 * Make the consumer endpoint as reactive independently of an input channel and
	 * apply the provided function into the {@link Flux#transform(Function)} operator.
	 * @param reactiveCustomizer the function to transform {@link Flux} for the input channel.
	 * @return the spec
	 * @since 5.5
	 */
	public S reactive(Function<? super Flux<Message<?>>, ? extends Publisher<Message<?>>> reactiveCustomizer) {

There you can set a retry() operator on the Flux.
The returned CompletableFuture from JmsOutboundGateway is converted to the Mono in this case: Mono.fromFuture() and that FluxMessageChannel is subscribed to this Mono. Therefore that's where we could retry async mode.

[LukaszPa]

I'm note sure that I understand this.

In my case i use JmsOutboundGateway for mq connection - inputChannel is ExecutorChannel.

Currently ServiceActivator don't have any output channel.
Gateway interface method returns CompletableFuture object.

---

@MessagingGateway
public interface SimpleGateway {
@gateway(requestChannel = "mqRequestChannel")
CompletableFuture sendAndReceiveMqMessage(String message);
}

---

configuration class

@bean
public MessageChannel mqRequestChannel(AsyncTaskExecutor taskExecutor) {
return new ExecutorChannel(taskExecutor);
}

@ServiceActivator(inputChannel = "mqRequestChannel")
@Bean
public JmsOutboundGateway jmsOutboundGateway(... ) {
    JmsOutboundGateway gateway = new JmsOutboundGateway();
    gateway.setConnectionFactory(mqConnectionFactory);
    gateway.setRequestDestinationName(requestDestination);
    gateway.setReplyDestinationName(responseDestination);
    gateway.setAsync(true);
     ....
     ....
    return gateway;
}

---

@Autowired
private SimpleGateway simpleGateway;

 CompletableFuture<String> mqResponse = simpleGateway.sendAndReceiveMqMessage("sdsadas");

---

I used ExecutorChannel to have control at threads number.

So with FluxMessageChannel it is possible handle exception from simpleGateway.sendAndReceiveMqMessage and retry it ?

As I understand simpleGateway.sendAndReceiveMqMessage result (CompletableFuture_) should be converted to Mono and then passed to FluxMessageChannel.
Next there should be ServiceActivator for fluxMessageChannel to process retry ?

I tried to find some examples, but unfortunately I didn't find anything.

[artembilan]

Looks like this is your SO thread: https://stackoverflow.com/questions/76968452/how-to-implement-thread-efficient-retries-on-async-jmsoutboundgateway.

I'll look into that next week when we come back from SpringOne.

[LukaszPa]

Yes, its my thread on SO. Thanks for help.
SpringOne great event - maybe someday I'll be able to participate. :)