diff --git a/build.gradle b/build.gradle index 3aaf951c..70a584d5 100644 --- a/build.gradle +++ b/build.gradle @@ -481,6 +481,7 @@ project('spring-rabbitmq-client') { api "com.rabbitmq.client:amqp-client:$rabbitmqAmqpClientVersion" testApi project(':spring-rabbit-junit') + testApi 'io.projectreactor:reactor-core' testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind' diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java index f49266e0..88e395d6 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java @@ -216,6 +216,10 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe beforeSendReplyPostProcessors.length); } + public MessagePostProcessor @Nullable [] getBeforeSendReplyPostProcessors() { + return this.beforeSendReplyPostProcessors; + } + /** * Set a {@link RetryTemplate} to use when sending replies. * @param retryTemplate the template. @@ -369,7 +373,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe /** * Handle the given result object returned from the listener method, sending a * response message back. - * @param resultArg the result object to handle (never null) + * @param resultArg the result object to handle * @param request the original request message * @param channel the Rabbit channel to operate on (maybe null) * @param source the source data for the method invocation - e.g. @@ -383,7 +387,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe protected void handleResult(@Nullable InvocationResult resultArg, Message request, @Nullable Channel channel, @Nullable Object source) { - if (channel != null && resultArg != null) { + if (resultArg != null) { if (resultArg.getReturnValue() instanceof CompletableFuture completable) { if (!this.isManualAck) { this.logger.warn("Container AcknowledgeMode must be MANUAL for a Future return type; " @@ -413,13 +417,9 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe doHandleResult(resultArg, request, channel, source); } } - else if (this.logger.isWarnEnabled()) { - this.logger.warn("Listener method returned result [" + resultArg - + "]: not generating response message for it because no Rabbit Channel given"); - } } - private void asyncSuccess(InvocationResult resultArg, Message request, Channel channel, + private void asyncSuccess(InvocationResult resultArg, Message request, @Nullable Channel channel, @Nullable Object source, @Nullable Object deferredResult) { if (deferredResult == null) { @@ -458,8 +458,9 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe } } - protected void asyncFailure(Message request, Channel channel, Throwable t, @Nullable Object source) { + protected void asyncFailure(Message request, @Nullable Channel channel, Throwable t, @Nullable Object source) { this.logger.error("Future, Mono, or suspend function was completed with an exception for " + request, t); + Assert.notNull(channel, "'channel' must not be null."); try { channel.basicNack(request.getMessageProperties().getDeliveryTag(), false, ContainerUtils.shouldRequeue(this.defaultRequeueRejected, t, this.logger)); @@ -469,7 +470,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe } } - protected void doHandleResult(InvocationResult resultArg, Message request, Channel channel, + protected void doHandleResult(InvocationResult resultArg, Message request, @Nullable Channel channel, @Nullable Object source) { if (this.logger.isDebugEnabled()) { @@ -500,12 +501,13 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe /** * Build a Rabbit message to be sent as response based on the given result object. * @param channel the Rabbit Channel to operate on. + * Can be null if implementation does not support AMQP 0.9.1. * @param result the content of the message, as returned from the listener method. * @param genericType the generic type to populate type headers. * @return the Rabbit Message (never null). * @see #setMessageConverter */ - protected Message buildMessage(Channel channel, @Nullable Object result, @Nullable Type genericType) { + protected Message buildMessage(@Nullable Channel channel, @Nullable Object result, @Nullable Type genericType) { MessageConverter converter = getMessageConverter(); if (converter != null && !(result instanceof Message)) { return convert(result, genericType, converter); @@ -633,7 +635,8 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe * @see #postProcessResponse(Message, Message) * @see #setReplyPostProcessor(ReplyPostProcessor) */ - protected void sendResponse(Channel channel, Address replyTo, Message messageIn) { + protected void sendResponse(@Nullable Channel channel, Address replyTo, Message messageIn) { + Assert.notNull(channel, "'channel' must not be null."); Message message = messageIn; if (this.beforeSendReplyPostProcessors != null) { for (MessagePostProcessor postProcessor : this.beforeSendReplyPostProcessors) { diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index a0a99834..20ce47ea 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java @@ -169,7 +169,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis } @Override - protected void asyncFailure(org.springframework.amqp.core.Message request, Channel channel, Throwable t, + protected void asyncFailure(org.springframework.amqp.core.Message request, @Nullable Channel channel, Throwable t, @Nullable Object source) { try { @@ -183,7 +183,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis super.asyncFailure(request, channel, t, source); } - private void handleException(org.springframework.amqp.core.Message amqpMessage, @Nullable Channel channel, + protected void handleException(org.springframework.amqp.core.Message amqpMessage, @Nullable Channel channel, @Nullable Message message, ListenerExecutionFailedException e) throws Exception { // NOSONAR if (this.errorHandler != null) { @@ -307,7 +307,7 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis * @see #setMessageConverter */ @Override - protected org.springframework.amqp.core.Message buildMessage(Channel channel, @Nullable Object result, + protected org.springframework.amqp.core.Message buildMessage(@Nullable Channel channel, @Nullable Object result, @Nullable Type genericType) { MessageConverter converter = getMessageConverter(); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java index 58b1c5d9..4fc1d3c4 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapterTests.java @@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import com.rabbitmq.client.Channel; +import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; @@ -33,7 +34,6 @@ import org.springframework.amqp.core.Address; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.support.SendRetryContextAccessor; -import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.aop.framework.ProxyFactory; import org.springframework.retry.RetryPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; @@ -67,8 +67,15 @@ public class MessageListenerAdapterTests { public void init() { this.messageProperties = new MessageProperties(); this.messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN); - this.adapter = new MessageListenerAdapter(); - this.adapter.setMessageConverter(new SimpleMessageConverter()); + this.adapter = new MessageListenerAdapter() { + + @Override + protected void doHandleResult(InvocationResult resultArg, Message request, @Nullable Channel channel, + @Nullable Object source) { + + } + + }; } @Test @@ -77,7 +84,7 @@ public class MessageListenerAdapterTests { @Override protected Object[] buildListenerArguments(Object extractedMessage, Channel channel, Message message) { - return new Object[] { extractedMessage, channel, message }; + return new Object[] {extractedMessage, channel, message}; } } @@ -131,7 +138,15 @@ public class MessageListenerAdapterTests { } } - this.adapter = new MessageListenerAdapter(new Delegate(), "myPojoMessageMethod"); + this.adapter = new MessageListenerAdapter(new Delegate(), "myPojoMessageMethod") { + + @Override + protected void doHandleResult(InvocationResult resultArg, Message request, @Nullable Channel channel, + @Nullable Object source) { + + } + + }; this.adapter.onMessage(new Message("foo".getBytes(), messageProperties), null); assertThat(called.get()).isTrue(); } @@ -146,7 +161,7 @@ public class MessageListenerAdapterTests { @Test public void testMappedListenerMethod() throws Exception { - Map map = new HashMap(); + Map map = new HashMap<>(); map.put("foo", "handle"); map.put("bar", "notDefinedOnInterface"); this.adapter.setDefaultListenerMethod("anotherHandle"); @@ -186,6 +201,7 @@ public class MessageListenerAdapterTests { @Test public void testReplyRetry() throws Exception { + this.adapter = new MessageListenerAdapter(); this.adapter.setDefaultListenerMethod("handle"); this.adapter.setDelegate(this.simpleService); RetryPolicy retryPolicy = new SimpleRetryPolicy(2); @@ -210,7 +226,7 @@ public class MessageListenerAdapterTests { this.adapter.onMessage(message, channel); assertThat(this.simpleService.called).isEqualTo("handle"); assertThat(replyMessage.get()).isNotNull(); - assertThat(new String(replyMessage.get().getBody())).isEqualTo("processedfoo"); + assertThat(new String(replyMessage.get().getBody())).isEqualTo("processed foo"); assertThat(replyAddress.get()).isNotNull(); assertThat(replyAddress.get().getExchangeName()).isEqualTo("foo"); assertThat(replyAddress.get().getRoutingKey()).isEqualTo("bar"); @@ -224,7 +240,7 @@ public class MessageListenerAdapterTests { @SuppressWarnings("unused") public CompletableFuture myPojoMessageMethod(String input) { CompletableFuture future = new CompletableFuture<>(); - future.complete("processed" + input); + future.complete("processed " + input); return future; } @@ -270,18 +286,18 @@ public class MessageListenerAdapterTests { @Override public String handle(String input) { called = "handle"; - return "processed" + input; + return "processed " + input; } @Override public String anotherHandle(String input) { called = "anotherHandle"; - return "processed" + input; + return "processed " + input; } public String notDefinedOnInterface(String input) { called = "notDefinedOnInterface"; - return "processed" + input; + return "processed " + input; } } diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java index 9ab6d8a6..1686051f 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerContainer.java @@ -99,6 +99,8 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer, Be private @Nullable MessageListener proxy; + private boolean asyncReplies; + private ErrorHandler errorHandler = new ConditionalRejectingErrorHandler(); private @Nullable Collection afterReceivePostProcessors; @@ -255,6 +257,10 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer, Be @Override public void setupMessageListener(MessageListener messageListener) { this.messageListener = messageListener; + this.asyncReplies = messageListener.isAsyncReplies(); + if (this.messageListener instanceof RabbitAmqpMessageListenerAdapter rabbitAmqpMessageListenerAdapter) { + rabbitAmqpMessageListenerAdapter.setConnectionFactory(this.connectionFactory); + } this.proxy = this.messageListener; if (!ObjectUtils.isEmpty(this.adviceChain)) { ProxyFactory factory = new ProxyFactory(messageListener); @@ -276,6 +282,11 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer, Be Assert.state(this.queues != null, "At least one queue has to be provided for consuming."); Assert.state(this.messageListener != null, "The 'messageListener' must be provided."); + if (this.asyncReplies && this.autoSettle) { + LOG.info("Enforce MANUAL settlement for async replies."); + this.autoSettle = false; + } + this.messageListener.containerAckMode(this.autoSettle ? AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL); if (this.messageListener instanceof RabbitAmqpMessageListenerAdapter adapter && this.afterReceivePostProcessors != null) { diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java index 0bd49bf6..123e1b40 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpMessageListenerAdapter.java @@ -20,19 +20,27 @@ import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.concurrent.CompletableFuture; +import com.rabbitmq.client.Channel; import com.rabbitmq.client.amqp.Consumer; import org.jspecify.annotations.Nullable; +import org.springframework.amqp.core.Address; import org.springframework.amqp.core.AmqpAcknowledgment; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.listener.adapter.InvocationResult; import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter; import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; +import org.springframework.amqp.rabbit.listener.support.ContainerUtils; import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException; +import org.springframework.amqp.rabbitmq.client.AmqpConnectionFactory; +import org.springframework.amqp.rabbitmq.client.RabbitAmqpTemplate; import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * A {@link MessagingMessageListenerAdapter} extension for the {@link RabbitAmqpMessageListener}. @@ -44,6 +52,11 @@ import org.springframework.messaging.support.GenericMessage; *
  • {@link Consumer.Context} - RabbitMQ AMQP client consumer settlement API.
    • *
  • {@link org.springframework.amqp.core.AmqpAcknowledgment} - Spring AMQP acknowledgment abstraction: delegates to the {@link Consumer.Context}
    • * + *

    + * This class reuses the {@link MessagingMessageListenerAdapter} as much as possible just to avoid duplication. + * The {@link Channel} abstraction from AMQP Client 0.9.1 is out use and present here just for API compatibility + * and to follow DRY principle. + * Can be reworked eventually, when this AMQP 1.0 client won't be based on {@code spring-rabbit} dependency. * * @author Artem Bilan * @@ -54,6 +67,8 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd private @Nullable Collection afterReceivePostProcessors; + private @Nullable RabbitAmqpTemplate rabbitAmqpTemplate; + public RabbitAmqpMessageListenerAdapter(@Nullable Object bean, @Nullable Method method, boolean returnExceptions, @Nullable RabbitListenerErrorHandler errorHandler, boolean batch) { @@ -64,6 +79,14 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd this.afterReceivePostProcessors = new ArrayList<>(afterReceivePostProcessors); } + /** + * Set a {@link AmqpConnectionFactory} for publishing replies from this adapter. + * @param connectionFactory the {@link AmqpConnectionFactory} for replies. + */ + public void setConnectionFactory(AmqpConnectionFactory connectionFactory) { + this.rabbitAmqpTemplate = new RabbitAmqpTemplate(connectionFactory); + } + @Override public void onAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage, Consumer.@Nullable Context context) { org.springframework.amqp.core.Message springMessage = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, context); @@ -78,15 +101,74 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd .invoke(messagingMessage, springMessage, springMessage.getMessageProperties().getAmqpAcknowledgment(), amqpMessage, context); + if (result.getReturnValue() != null) { - logger.warn("Replies are not currently supported with RabbitMQ AMQP 1.0 listeners"); + Assert.notNull(this.rabbitAmqpTemplate, + "The 'connectionFactory' must be provided for handling replies."); + handleResult(result, springMessage, null, messagingMessage); } + } catch (Exception ex) { throw new ListenerExecutionFailedException("Failed to invoke listener", ex, springMessage); } } + @Override + protected void asyncFailure(Message request, @Nullable Channel channel, Throwable t, @Nullable Object source) { + try { + handleException(request, channel, (org.springframework.messaging.Message) source, + new ListenerExecutionFailedException("Async Fail", t, request)); + return; + } + catch (Exception ex) { + // Ignore and reject the message against original error + } + + this.logger.error("Future, Mono, or suspend function was completed with an exception for " + request, t); + AmqpAcknowledgment amqpAcknowledgment = request.getMessageProperties().getAmqpAcknowledgment(); + Assert.notNull(amqpAcknowledgment, "'(amqpAcknowledgment' must be provided into request message."); + + if (ContainerUtils.shouldRequeue(isDefaultRequeueRejected(), t, this.logger)) { + amqpAcknowledgment.acknowledge(AmqpAcknowledgment.Status.REQUEUE); + } + else { + amqpAcknowledgment.acknowledge(AmqpAcknowledgment.Status.REJECT); + } + } + + @Override + @SuppressWarnings("NullAway") // Dataflow analysis limitation + protected void sendResponse(@Nullable Channel channel, Address replyTo, Message messageIn) { + Message replyMessage = messageIn; + MessagePostProcessor[] beforeSendReplyPostProcessors = getBeforeSendReplyPostProcessors(); + if (beforeSendReplyPostProcessors != null) { + for (MessagePostProcessor postProcessor : beforeSendReplyPostProcessors) { + replyMessage = postProcessor.postProcessMessage(replyMessage); + } + } + + String replyToExchange = replyTo.getExchangeName(); + String replyToRoutingKey = replyTo.getRoutingKey(); + CompletableFuture sendFuture; + if (StringUtils.hasText(replyToExchange)) { + sendFuture = this.rabbitAmqpTemplate.send(replyToExchange, replyToRoutingKey, replyMessage); + } + else { + Assert.hasText(replyToRoutingKey, "The 'replyTo' must be provided, in request message or in @SendTo."); + sendFuture = this.rabbitAmqpTemplate.send(replyToRoutingKey.replaceFirst("queues/", ""), replyMessage); + } + + sendFuture.join(); + } + + @Override + protected void basicAck(Message request, @Nullable Channel channel) { + AmqpAcknowledgment amqpAcknowledgment = request.getMessageProperties().getAmqpAcknowledgment(); + Assert.notNull(amqpAcknowledgment, "'(amqpAcknowledgment' must be provided into request message."); + amqpAcknowledgment.acknowledge(); + } + @Override public void onMessageBatch(List messages) { AmqpAcknowledgment amqpAcknowledgment = diff --git a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java index 7d286157..17d68e17 100644 --- a/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java +++ b/spring-rabbitmq-client/src/test/java/org/springframework/amqp/rabbitmq/client/listener/RabbitAmqpListenerTests.java @@ -28,8 +28,12 @@ import java.util.stream.IntStream; import com.rabbitmq.client.amqp.Consumer; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; import org.springframework.amqp.core.AmqpAcknowledgment; +import org.springframework.amqp.core.Binding; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.rabbit.annotation.EnableRabbit; @@ -45,6 +49,7 @@ import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; import org.springframework.util.MultiValueMap; @@ -144,6 +149,32 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase { assertThat(this.config.batchReceivedOnThread).startsWith("dispatching-rabbitmq-amqp-"); } + @Test + void verifyBasicRequestReply() { + CompletableFuture replyFuture = this.template.convertSendAndReceive("requestQueue", "test data"); + assertThat(replyFuture).succeedsWithin(10, TimeUnit.SECONDS).isEqualTo("TEST DATA"); + } + + @Test + void verifyFutureReturnRequestReply() { + CompletableFuture replyFuture = this.template.convertSendAndReceive("requestQueue2", "TEST DATA2"); + assertThat(replyFuture).succeedsWithin(10, TimeUnit.SECONDS).isEqualTo("test data2"); + } + + @Test + void verifyMonoReturnRequestReply() { + CompletableFuture replyFuture = this.template.convertSendAndReceive("requestQueue3", "test data3"); + assertThat(replyFuture).succeedsWithin(10, TimeUnit.SECONDS).isEqualTo("Mono test data3"); + } + + @Test + void verifyReplyOnAnotherQueue() { + this.template.convertAndSend("requestQueue4", "test data4"); + CompletableFuture replyFuture = this.template.receiveAndConvert("q4"); + assertThat(replyFuture).succeedsWithin(10, TimeUnit.SECONDS) + .isEqualTo("Reply for 'test data4' via 'e1' and 'k4'"); + } + @Configuration @EnableRabbit static class Config { @@ -163,6 +194,21 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase { return new Queue("q3"); } + @Bean + DirectExchange e1() { + return new DirectExchange("e1"); + } + + @Bean + Queue q4() { + return new Queue("q4"); + } + + @Bean + Binding b4() { + return BindingBuilder.bind(q4()).to(e1()).with("k4"); + } + @Bean(RabbitListenerAnnotationBeanPostProcessor.DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME) RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory(AmqpConnectionFactory connectionFactory) { return new RabbitAmqpListenerContainerFactory(connectionFactory); @@ -231,6 +277,29 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase { this.batchReceived.complete(data); } + @RabbitListener(queuesToDeclare = @org.springframework.amqp.rabbit.annotation.Queue("requestQueue")) + String toUpperCaseRpc(String data) { + return data.toUpperCase(); + } + + @RabbitListener(queuesToDeclare = @org.springframework.amqp.rabbit.annotation.Queue("requestQueue2")) + CompletableFuture toLowerCaseFutureRpc(String data) { + return CompletableFuture.completedFuture(data) + .thenApply(String::toLowerCase); + } + + @RabbitListener(queuesToDeclare = @org.springframework.amqp.rabbit.annotation.Queue("requestQueue3")) + Mono monoRpc(String data) { + return Mono.just(data) + .map(value -> "Mono " + value); + } + + @RabbitListener(queuesToDeclare = @org.springframework.amqp.rabbit.annotation.Queue("requestQueue4")) + @SendTo("e1/k4") + String replyViaSendTo(String data) { + return "Reply for '%s' via 'e1' and 'k4'".formatted(data); + } + } }