diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapter.java index ecb97f60..144d064b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/BatchMessagingMessageListenerAdapter.java @@ -49,8 +49,6 @@ import org.springframework.util.Assert; public class BatchMessagingMessageListenerAdapter extends MessagingMessageListenerAdapter implements ChannelAwareBatchMessageListener { - private final MessagingMessageConverterAdapter converterAdapter; - private final BatchingStrategy batchingStrategy; @SuppressWarnings("this-escape") @@ -58,14 +56,13 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen @Nullable RabbitListenerErrorHandler errorHandler, @Nullable BatchingStrategy batchingStrategy) { super(bean, method, returnExceptions, errorHandler, true); - this.converterAdapter = (MessagingMessageConverterAdapter) getMessagingMessageConverter(); this.batchingStrategy = batchingStrategy == null ? new SimpleBatchingStrategy(0, 0, 0L) : batchingStrategy; } @Override public void onMessageBatch(List messages, @Nullable Channel channel) { Message converted; - if (this.converterAdapter.isAmqpMessageList()) { + if (this.messagingMessageConverter.isAmqpMessageList()) { converted = new GenericMessage<>(messages); } else { @@ -87,7 +84,7 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen } } } - if (this.converterAdapter.isMessageList()) { + if (this.messagingMessageConverter.isMessageList()) { converted = new GenericMessage<>(messagingMessages); } else { @@ -178,7 +175,7 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen protected Message toMessagingMessage(org.springframework.amqp.core.Message amqpMessage) { if (this.batchingStrategy.canDebatch(amqpMessage.getMessageProperties())) { - if (this.converterAdapter.isMessageList()) { + if (this.messagingMessageConverter.isMessageList()) { List> messages = new ArrayList<>(); this.batchingStrategy.deBatch(amqpMessage, fragment -> messages.add(super.toMessagingMessage(fragment))); return new GenericMessage<>(messages); @@ -186,9 +183,9 @@ public class BatchMessagingMessageListenerAdapter extends MessagingMessageListen else { List list = new ArrayList<>(); this.batchingStrategy.deBatch(amqpMessage, fragment -> - list.add(this.converterAdapter.extractPayload(fragment))); + list.add(this.messagingMessageConverter.extractPayload(fragment))); return MessageBuilder.withPayload(list) - .copyHeaders(this.converterAdapter + .copyHeaders(this.messagingMessageConverter .getHeaderMapper() .toHeaders(amqpMessage.getMessageProperties())) .build(); 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 b6033b14..a0a99834 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 @@ -69,11 +69,11 @@ import org.springframework.util.TypeUtils; */ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageListener { - private final MessagingMessageConverterAdapter messagingMessageConverter; + protected final MessagingMessageConverterAdapter messagingMessageConverter; - private final boolean returnExceptions; + protected final boolean returnExceptions; - private final @Nullable RabbitListenerErrorHandler errorHandler; + protected final @Nullable RabbitListenerErrorHandler errorHandler; private @Nullable HandlerAdapter handlerAdapter; @@ -364,15 +364,15 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis } } - protected boolean isMessageList() { + public boolean isMessageList() { return this.isMessageList; } - protected boolean isAmqpMessageList() { + public boolean isAmqpMessageList() { return this.isAmqpMessageList; } - protected @Nullable Method getMethod() { + public @Nullable Method getMethod() { return this.method; } diff --git a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java index 926580f5..6195a216 100644 --- a/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java +++ b/spring-rabbitmq-client/src/main/java/org/springframework/amqp/rabbitmq/client/config/RabbitAmqpListenerContainerFactory.java @@ -16,10 +16,13 @@ package org.springframework.amqp.rabbitmq.client.config; +import java.util.Arrays; + import com.rabbitmq.client.amqp.Connection; -import org.aopalliance.aop.Advice; import org.jspecify.annotations.Nullable; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.config.ContainerCustomizer; import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint; @@ -27,6 +30,7 @@ import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpListenerContainer; import org.springframework.amqp.rabbitmq.client.listener.RabbitAmqpMessageListenerAdapter; import org.springframework.amqp.utils.JavaUtils; +import org.springframework.scheduling.TaskScheduler; /** * Factory for {@link RabbitAmqpListenerContainer}. @@ -45,6 +49,14 @@ public class RabbitAmqpListenerContainerFactory private @Nullable ContainerCustomizer containerCustomizer; + private MessagePostProcessor @Nullable [] afterReceivePostProcessors; + + private @Nullable Integer batchSize; + + private @Nullable Long batchReceiveTimeout; + + private @Nullable TaskScheduler taskScheduler; + /** * Construct an instance using the provided amqpConnection. * @param amqpConnection the connection. @@ -62,18 +74,69 @@ public class RabbitAmqpListenerContainerFactory this.containerCustomizer = containerCustomizer; } + /** + * Set {@link MessagePostProcessor}s that will be applied after message reception, before + * invoking the {@link MessageListener}. Often used to decompress data. Processors are invoked in order, + * depending on {@code PriorityOrder}, {@code Order} and finally unordered. + * @param afterReceivePostProcessors the post processor. + */ + public void setAfterReceivePostProcessors(MessagePostProcessor... afterReceivePostProcessors) { + this.afterReceivePostProcessors = Arrays.copyOf(afterReceivePostProcessors, afterReceivePostProcessors.length); + } + + /** + * The size of the batch of messages to process. + * This is only option (if {@code batchSize > 1}) which turns the target listener container into a batch mode. + * @param batchSize the batch size. + * @see RabbitAmqpListenerContainer#setBatchSize + * @see #setBatchReceiveTimeout(Long) + */ + public void setBatchSize(Integer batchSize) { + this.batchSize = batchSize; + } + + /** + * The number of milliseconds of timeout for gathering batch messages. + * It limits the time to wait to fill batchSize. + * Default is 30 seconds. + * @param batchReceiveTimeout the timeout for gathering batch messages. + * @see RabbitAmqpListenerContainer#setBatchReceiveTimeout + * @see #setBatchSize(Integer) + */ + public void setBatchReceiveTimeout(Long batchReceiveTimeout) { + this.batchReceiveTimeout = batchReceiveTimeout; + } + + /** + * Configure a {@link TaskScheduler} to release not fulfilled batches after timeout. + * @param taskScheduler the {@link TaskScheduler} to use. + * @see RabbitAmqpListenerContainer#setTaskScheduler(TaskScheduler) + * @see #setBatchReceiveTimeout(Long) + */ + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + } + @Override public RabbitAmqpListenerContainer createListenerContainer(@Nullable RabbitListenerEndpoint endpoint) { if (endpoint instanceof MethodRabbitListenerEndpoint methodRabbitListenerEndpoint) { + JavaUtils.INSTANCE + .acceptIfCondition(this.batchSize != null && this.batchSize > 1, + true, + methodRabbitListenerEndpoint::setBatchListener); + methodRabbitListenerEndpoint.setAdapterProvider( (batch, bean, method, returnExceptions, errorHandler, batchingStrategy) -> - new RabbitAmqpMessageListenerAdapter(bean, method, returnExceptions, errorHandler)); + new RabbitAmqpMessageListenerAdapter(bean, method, returnExceptions, errorHandler, batch)); } RabbitAmqpListenerContainer container = createContainerInstance(); - Advice[] adviceChain = getAdviceChain(); JavaUtils.INSTANCE - .acceptIfNotNull(adviceChain, container::setAdviceChain) - .acceptIfNotNull(getDefaultRequeueRejected(), container::setDefaultRequeue); + .acceptIfNotNull(getAdviceChain(), container::setAdviceChain) + .acceptIfNotNull(getDefaultRequeueRejected(), container::setDefaultRequeue) + .acceptIfNotNull(this.afterReceivePostProcessors, container::setAfterReceivePostProcessors) + .acceptIfNotNull(this.batchSize, container::setBatchSize) + .acceptIfNotNull(this.batchReceiveTimeout, container::setBatchReceiveTimeout) + .acceptIfNotNull(this.taskScheduler, container::setTaskScheduler); applyCommonOverrides(endpoint, container); 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 034dd85a..eac16999 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 @@ -17,9 +17,14 @@ package org.springframework.amqp.rabbitmq.client.listener; import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.List; +import java.util.Objects; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -36,13 +41,19 @@ import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AmqpAcknowledgment; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.support.ContainerUtils; import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils; +import org.springframework.amqp.support.postprocessor.MessagePostProcessorUtils; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.DefaultPointcutAdvisor; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.DisposableBean; import org.springframework.core.log.LogAccessor; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; import org.springframework.util.LinkedMultiValueMap; @@ -57,7 +68,7 @@ import org.springframework.util.ObjectUtils; * @since 4.0 * */ -public class RabbitAmqpListenerContainer implements MessageListenerContainer { +public class RabbitAmqpListenerContainer implements MessageListenerContainer, BeanNameAware, DisposableBean { private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(RabbitAmqpListenerContainer.class)); @@ -85,14 +96,28 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { private @Nullable MessageListener messageListener; + private @Nullable MessageListener proxy; + private ErrorHandler errorHandler = new ConditionalRejectingErrorHandler(); + private @Nullable Collection afterReceivePostProcessors; + private boolean autoStartup = true; + private String beanName = "not.a.Spring.bean"; + private @Nullable String listenerId; private Duration gracefulShutdownPeriod = Duration.ofSeconds(30); + private int batchSize; + + private Duration batchReceiveDuration = Duration.ofSeconds(30); + + private @Nullable TaskScheduler taskScheduler; + + private boolean internalTaskScheduler = true; + /** * Construct an instance using the provided connection. * @param connection to use. @@ -118,6 +143,39 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { this.stateListeners = Arrays.copyOf(stateListeners, stateListeners.length); } + /** + * Set {@link MessagePostProcessor}s that will be applied after message reception, before + * invoking the {@link MessageListener}. Often used to decompress data. Processors are invoked in order, + * depending on {@code PriorityOrder}, {@code Order} and finally unordered. + * @param afterReceivePostProcessors the post processor. + */ + public void setAfterReceivePostProcessors(MessagePostProcessor... afterReceivePostProcessors) { + this.afterReceivePostProcessors = MessagePostProcessorUtils.sort(Arrays.asList(afterReceivePostProcessors)); + } + + public void setBatchSize(int batchSize) { + Assert.isTrue(batchSize > 1, "'batchSize' must be greater than 1"); + this.batchSize = batchSize; + } + + public void setBatchReceiveTimeout(long batchReceiveTimeout) { + this.batchReceiveDuration = Duration.ofMillis(batchReceiveTimeout); + } + + /** + * Set a {@link TaskScheduler} for monitoring batch releases. + * @param taskScheduler the {@link TaskScheduler} to use. + */ + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + this.internalTaskScheduler = false; + } + + @Override + public void setBeanName(String name) { + this.beanName = name; + } + @Override public void setAutoStartup(boolean autoStart) { this.autoStartup = autoStart; @@ -185,22 +243,31 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { this.listenerId = id; } + /** + * The 'id' attribute of the listener. + * @return the id (or the container bean name if no id set). + */ + public String getListenerId() { + return this.listenerId != null ? this.listenerId : this.beanName; + } + @Override public void setupMessageListener(MessageListener messageListener) { this.messageListener = messageListener; + this.proxy = this.messageListener; if (!ObjectUtils.isEmpty(this.adviceChain)) { ProxyFactory factory = new ProxyFactory(messageListener); for (Advice advice : this.adviceChain) { factory.addAdvisor(new DefaultPointcutAdvisor(advice)); } factory.setInterfaces(messageListener.getClass().getInterfaces()); - this.messageListener = (MessageListener) factory.getProxy(getClass().getClassLoader()); + this.proxy = (MessageListener) factory.getProxy(getClass().getClassLoader()); } } @Override public @Nullable Object getMessageListener() { - return this.messageListener; + return this.proxy; } @Override @@ -209,6 +276,18 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { Assert.state(this.messageListener != null, "The 'messageListener' must be provided."); this.messageListener.containerAckMode(this.autoSettle ? AcknowledgeMode.AUTO : AcknowledgeMode.MANUAL); + if (this.messageListener instanceof RabbitAmqpMessageListenerAdapter adapter + && this.afterReceivePostProcessors != null) { + + adapter.setAfterReceivePostProcessors(this.afterReceivePostProcessors); + } + + if (this.batchSize > 1 && this.internalTaskScheduler) { + ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); + threadPoolTaskScheduler.setThreadNamePrefix(getListenerId() + "-consumerMonitor-"); + threadPoolTaskScheduler.afterPropertiesSet(); + this.taskScheduler = threadPoolTaskScheduler; + } } @Override @@ -236,7 +315,7 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { .priority(this.priority) .initialCredits(this.initialCredits) .listeners(this.stateListeners) - .messageHandler(this::invokeListener) + .messageHandler(new ConsumerMessageHandler()) .build(); this.queueToConsumers.add(queue, consumer); } @@ -256,39 +335,65 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { } } catch (Exception ex) { - try { - this.errorHandler.handleError(ex); - // If error handler does not re-throw an exception, re-check original error. - // If it is not special, treat the error handler outcome as a successful processing result. - if (!handleSpecialErrors(ex, context)) { - context.accept(); - } - } - catch (Exception rethrow) { - if (!handleSpecialErrors(rethrow, context)) { - if (this.defaultRequeue) { - context.requeue(); - } - else { - context.discard(); - } - LOG.error(rethrow, () -> - "The 'errorHandler' has thrown an exception. The '" + amqpMessage + "' is " - + (this.defaultRequeue ? "re-queued." : "discarded.")); - } - } + handleListenerError(ex, context, amqpMessage); } } @SuppressWarnings("NullAway") // Dataflow analysis limitation private void doInvokeListener(Consumer.Context context, com.rabbitmq.client.amqp.Message amqpMessage) { Consumer.@Nullable Context contextToUse = this.autoSettle ? null : context; - if (this.messageListener instanceof RabbitAmqpMessageListener amqpMessageListener) { + if (this.proxy instanceof RabbitAmqpMessageListener amqpMessageListener) { amqpMessageListener.onAmqpMessage(amqpMessage, contextToUse); } else { Message message = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, contextToUse); - this.messageListener.onMessage(message); + this.proxy.onMessage(message); + } + } + + private void invokeBatchListener(Consumer.Context context, List batch) { + Consumer.@Nullable Context contextToUse = this.autoSettle ? null : context; + List messages = + batch.stream() + .map((amqpMessage) -> RabbitAmqpUtils.fromAmqpMessage(amqpMessage, contextToUse)) + .toList(); + try { + doInvokeBatchListener(messages); + if (this.autoSettle) { + context.accept(); + } + } + catch (Exception ex) { + handleListenerError(ex, context, batch); + } + } + + @SuppressWarnings("NullAway") // Dataflow analysis limitation + private void doInvokeBatchListener(List messages) { + this.proxy.onMessageBatch(messages); + } + + private void handleListenerError(Exception ex, Consumer.Context context, Object messageOrBatch) { + try { + this.errorHandler.handleError(ex); + // If error handler does not re-throw an exception, re-check original error. + // If it is not special, treat the error handler outcome as a successful processing result. + if (!handleSpecialErrors(ex, context)) { + context.accept(); + } + } + catch (Exception rethrow) { + if (!handleSpecialErrors(rethrow, context)) { + if (this.defaultRequeue) { + context.requeue(); + } + else { + context.discard(); + } + LOG.error(rethrow, () -> + "The 'errorHandler' has thrown an exception. The '" + messageOrBatch + "' is " + + (this.defaultRequeue ? "re-queued." : "discarded.")); + } } } @@ -390,4 +495,79 @@ public class RabbitAmqpListenerContainer implements MessageListenerContainer { } } + @Override + public void destroy() { + if (this.internalTaskScheduler && this.taskScheduler != null) { + ((ThreadPoolTaskScheduler) this.taskScheduler).shutdown(); + } + } + + private class ConsumerMessageHandler implements Consumer.MessageHandler { + + private volatile @Nullable ConsumerBatch consumerBatch; + + ConsumerMessageHandler() { + } + + @Override + public void handle(Consumer.Context context, com.rabbitmq.client.amqp.Message message) { + if (RabbitAmqpListenerContainer.this.batchSize > 1) { + ConsumerBatch currentBatch = this.consumerBatch; + if (currentBatch == null || currentBatch.batchReleaseFuture == null) { + currentBatch = new ConsumerBatch(context.batch(RabbitAmqpListenerContainer.this.batchSize)); + this.consumerBatch = currentBatch; + } + currentBatch.add(context, message); + if (currentBatch.batchContext.size() == RabbitAmqpListenerContainer.this.batchSize) { + currentBatch.release(); + this.consumerBatch = null; + } + } + else { + invokeListener(context, message); + } + } + + private class ConsumerBatch { + + private final List batch = new ArrayList<>(); + + private final Consumer.BatchContext batchContext; + + private volatile @Nullable ScheduledFuture batchReleaseFuture; + + ConsumerBatch(Consumer.BatchContext batchContext) { + this.batchContext = batchContext; + } + + void add(Consumer.Context context, com.rabbitmq.client.amqp.Message message) { + this.batchContext.add(context); + this.batch.add(message); + if (this.batchReleaseFuture == null) { + this.batchReleaseFuture = + Objects.requireNonNull(RabbitAmqpListenerContainer.this.taskScheduler) + .schedule(this::releaseInternal, + Instant.now().plus(RabbitAmqpListenerContainer.this.batchReceiveDuration)); + } + } + + void release() { + ScheduledFuture currentBatchReleaseFuture = this.batchReleaseFuture; + if (currentBatchReleaseFuture != null) { + currentBatchReleaseFuture.cancel(true); + releaseInternal(); + } + } + + private void releaseInternal() { + if (this.batchReleaseFuture != null) { + this.batchReleaseFuture = null; + invokeBatchListener(this.batchContext, this.batch); + } + } + + } + + } + } 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 d52f0d03..a94e2f88 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 @@ -17,15 +17,22 @@ package org.springframework.amqp.rabbitmq.client.listener; import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; import com.rabbitmq.client.amqp.Consumer; import org.jspecify.annotations.Nullable; +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.support.ListenerExecutionFailedException; import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils; +import org.springframework.messaging.support.GenericMessage; /** * A {@link MessagingMessageListenerAdapter} extension for the {@link RabbitAmqpMessageListener}. @@ -45,15 +52,26 @@ import org.springframework.amqp.rabbitmq.client.RabbitAmqpUtils; public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAdapter implements RabbitAmqpMessageListener { - public RabbitAmqpMessageListenerAdapter(@Nullable Object bean, @Nullable Method method, boolean returnExceptions, - @Nullable RabbitListenerErrorHandler errorHandler) { + private @Nullable Collection afterReceivePostProcessors; - super(bean, method, returnExceptions, errorHandler); + public RabbitAmqpMessageListenerAdapter(@Nullable Object bean, @Nullable Method method, boolean returnExceptions, + @Nullable RabbitListenerErrorHandler errorHandler, boolean batch) { + + super(bean, method, returnExceptions, errorHandler, batch); + } + + public void setAfterReceivePostProcessors(Collection afterReceivePostProcessors) { + this.afterReceivePostProcessors = new ArrayList<>(afterReceivePostProcessors); } @Override public void onAmqpMessage(com.rabbitmq.client.amqp.Message amqpMessage, Consumer.@Nullable Context context) { org.springframework.amqp.core.Message springMessage = RabbitAmqpUtils.fromAmqpMessage(amqpMessage, context); + if (this.afterReceivePostProcessors != null) { + for (MessagePostProcessor processor : this.afterReceivePostProcessors) { + springMessage = processor.postProcessMessage(springMessage); + } + } try { org.springframework.messaging.Message messagingMessage = toMessagingMessage(springMessage); InvocationResult result = getHandlerAdapter() @@ -69,4 +87,46 @@ public class RabbitAmqpMessageListenerAdapter extends MessagingMessageListenerAd } } + @Override + public void onMessageBatch(List messages) { + AmqpAcknowledgment amqpAcknowledgment = + messages.stream() + .findAny() + .map((message) -> message.getMessageProperties().getAmqpAcknowledgment()) + .orElse(null); + + org.springframework.messaging.Message converted; + if (this.messagingMessageConverter.isAmqpMessageList()) { + converted = new GenericMessage<>(messages); + } + else { + List> messagingMessages = + messages.stream() + .map(this::toMessagingMessage) + .toList(); + + if (this.messagingMessageConverter.isMessageList()) { + converted = new GenericMessage<>(messagingMessages); + } + else { + List payloads = new ArrayList<>(); + for (org.springframework.messaging.Message message : messagingMessages) { + payloads.add(message.getPayload()); + } + converted = new GenericMessage<>(payloads); + } + } + try { + InvocationResult result = getHandlerAdapter() + .invoke(converted, amqpAcknowledgment); + if (result.getReturnValue() != null) { + logger.warn("Replies are not currently supported with RabbitMQ AMQP 1.0 listeners"); + } + } + catch (Exception ex) { + throw new ListenerExecutionFailedException("Failed to invoke listener", ex, + messages.toArray(new Message[0])); + } + } + } 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 55d83b29..22af681d 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 @@ -20,8 +20,10 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Random; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; import com.rabbitmq.client.amqp.Connection; import com.rabbitmq.client.amqp.Consumer; @@ -42,10 +44,13 @@ import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry; import org.springframework.amqp.rabbitmq.client.RabbitAmqpTestBase; import org.springframework.amqp.rabbitmq.client.config.RabbitAmqpListenerContainerFactory; import org.springframework.amqp.support.converter.MessageConversionException; +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.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.test.context.ContextConfiguration; +import org.springframework.util.MultiValueMap; import static org.assertj.core.api.Assertions.assertThat; @@ -93,6 +98,53 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase { assertThat(this.template.receive("dlq1")).succeedsWithin(10, TimeUnit.SECONDS); } + @Test + @SuppressWarnings("unchecked") + void verifyBatchConsumedAfterScheduledTimeout() { + List testDataList = + List.of("batchData1", "batchData2", "batchData3", "batchData4", "batchData5"); + + for (String testData : testDataList) { + this.template.convertAndSend("q3", testData); + } + + assertThat(this.config.batchReceived).succeedsWithin(10, TimeUnit.SECONDS) + .asInstanceOf(InstanceOfAssertFactories.LIST) + .hasSize(5) + .containsAll(testDataList); + + assertThat(this.config.batchReceivedOnThread).startsWith("batch-consumer-scheduler-"); + + MessageListenerContainer testBatchListener = + this.rabbitListenerEndpointRegistry.getListenerContainer("testBatchListener"); + + MultiValueMap queueToConsumers = + TestUtils.getPropertyValue(testBatchListener, "queueToConsumers", MultiValueMap.class); + Consumer consumer = queueToConsumers.get("q3").get(0); + + assertThat(consumer.unsettledMessageCount()).isEqualTo(0L); + + this.config.batchReceived = new CompletableFuture<>(); + + testDataList = + IntStream.range(6, 16) + .boxed() + .map(Object::toString) + .map("batchData"::concat) + .toList(); + + for (String testData : testDataList) { + this.template.convertAndSend("q3", testData); + } + + assertThat(this.config.batchReceived).succeedsWithin(10, TimeUnit.SECONDS) + .asInstanceOf(InstanceOfAssertFactories.LIST) + .hasSize(10) + .containsAll(testDataList); + + assertThat(this.config.batchReceivedOnThread).startsWith("dispatching-rabbitmq-amqp-"); + } + @Configuration @EnableRabbit static class Config { @@ -122,6 +174,11 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase { return QueueBuilder.durable("q2").deadLetterExchange("dlx1").build(); } + @Bean + Queue q3() { + return new Queue("q3"); + } + @Bean(RabbitListenerAnnotationBeanPostProcessor.DEFAULT_RABBIT_LISTENER_CONTAINER_FACTORY_BEAN_NAME) RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory(Connection connection) { return new RabbitAmqpListenerContainerFactory(connection); @@ -158,6 +215,38 @@ class RabbitAmqpListenerTests extends RabbitAmqpTestBase { } } + @Bean + ThreadPoolTaskScheduler taskScheduler() { + ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); + threadPoolTaskScheduler.setPoolSize(2); + threadPoolTaskScheduler.setThreadNamePrefix("batch-consumer-scheduler-"); + return threadPoolTaskScheduler; + } + + @Bean + RabbitAmqpListenerContainerFactory batchRabbitAmqpListenerContainerFactory(Connection connection, + ThreadPoolTaskScheduler taskScheduler) { + + RabbitAmqpListenerContainerFactory rabbitAmqpListenerContainerFactory = + new RabbitAmqpListenerContainerFactory(connection); + rabbitAmqpListenerContainerFactory.setTaskScheduler(taskScheduler); + rabbitAmqpListenerContainerFactory.setBatchSize(10); + rabbitAmqpListenerContainerFactory.setBatchReceiveTimeout(1000L); + return rabbitAmqpListenerContainerFactory; + } + + CompletableFuture> batchReceived = new CompletableFuture<>(); + + volatile String batchReceivedOnThread; + + @RabbitListener(queues = "q3", + containerFactory = "batchRabbitAmqpListenerContainerFactory", + id = "testBatchListener") + void processBatchFromQ3(List data) { + this.batchReceivedOnThread = Thread.currentThread().getName(); + this.batchReceived.complete(data); + } + } }