GH-1484: RabbitListener.batch() Override

Resolves https://github.com/spring-projects/spring-amqp/issues/1484
This commit is contained in:
Gary Russell
2022-08-02 17:46:22 -04:00
committed by Artem Bilan
parent 97a508e379
commit 515eb9a32b
7 changed files with 52 additions and 10 deletions

View File

@@ -332,4 +332,18 @@ public @interface RabbitListener {
*/
String converterWinsContentType() default "true";
/**
* Override the container factory's {@code batchListener} property. The listener
* method signature should receive a {@code List<?>}; refer to the reference
* documentation. This allows a single container factory to be used for both record
* and batch listeners; previously separate container factories were required.
* @return "true" for the annotated method to be a batch listener or "false" for a
* single message listener. If not set, the container factory setting is used. SpEL and
* property place holders are not supported because the listener type cannot be
* variable.
* @since 3.0
* @see Boolean#parseBoolean(String)
*/
String batch() default "";
}

View File

@@ -461,6 +461,9 @@ public class RabbitListenerAnnotationBeanPostProcessor
resolvePostProcessor(endpoint, rabbitListener, target, beanName);
resolveMessageConverter(endpoint, rabbitListener, target, beanName);
resolveReplyContentType(endpoint, rabbitListener);
if (StringUtils.hasText(rabbitListener.batch())) {
endpoint.setBatchListener(Boolean.parseBoolean(rabbitListener.batch()));
}
RabbitListenerContainerFactory<?> factory = resolveContainerFactory(rabbitListener, target, beanName);
this.registrar.registerEndpoint(endpoint, factory);

View File

@@ -381,7 +381,9 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
.acceptIfNotNull(endpoint.getAckMode(), instance::setAcknowledgeMode)
.acceptIfNotNull(endpoint.getBatchingStrategy(), instance::setBatchingStrategy);
instance.setListenerId(endpoint.getId());
endpoint.setBatchListener(this.batchListener);
if (endpoint.getBatchListener() == null) {
endpoint.setBatchListener(this.batchListener);
}
}
applyCommonOverrides(endpoint, instance);

View File

@@ -84,7 +84,7 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
private TaskExecutor taskExecutor;
private boolean batchListener;
private Boolean batchListener;
private BatchingStrategy batchingStrategy;
@@ -293,7 +293,21 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn
this.taskExecutor = taskExecutor;
}
/**
* True if this endpoint is for a batch listener.
* @return true if batch.
*/
public boolean isBatchListener() {
return this.batchListener == null ? false : this.batchListener;
}
/**
* True if this endpoint is for a batch listener.
* @return {@link Boolean#TRUE} if batch.
* @since 3.0
*/
@Nullable
public Boolean getBatchListener() {
return this.batchListener;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2020 the original author or authors.
* Copyright 2014-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -130,7 +130,7 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
protected MessagingMessageListenerAdapter createMessageListener(MessageListenerContainer container) {
Assert.state(this.messageHandlerMethodFactory != null,
"Could not create message listener - MessageHandlerMethodFactory not set");
MessagingMessageListenerAdapter messageListener = createMessageListenerInstance();
MessagingMessageListenerAdapter messageListener = createMessageListenerInstance(getBatchListener());
messageListener.setHandlerAdapter(configureListenerAdapter(messageListener));
String replyToAddress = getDefaultReplyToAddress();
if (replyToAddress != null) {
@@ -159,11 +159,12 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
/**
* Create an empty {@link MessagingMessageListenerAdapter} instance.
* @param batch whether this endpoint is for a batch listener.
* @return the {@link MessagingMessageListenerAdapter} instance.
*/
protected MessagingMessageListenerAdapter createMessageListenerInstance() {
return this.adapterProvider.getAdapter(isBatchListener(), this.bean, this.method, this.returnExceptions,
this.errorHandler, getBatchingStrategy());
protected MessagingMessageListenerAdapter createMessageListenerInstance(@Nullable Boolean batch) {
return this.adapterProvider.getAdapter(batch == null ? isBatchListener() : batch, this.bean, this.method,
this.returnExceptions, this.errorHandler, getBatchingStrategy());
}
@Nullable

View File

@@ -116,6 +116,14 @@ public interface RabbitListenerEndpoint {
default void setBatchListener(boolean batchListener) {
}
/**
* Whether this endpoint is for a batch listener.
* @return {@link Boolean#TRUE} if batch.
* @since 3.0
*/
@Nullable
Boolean getBatchListener();
/**
* Set a {@link BatchingStrategy} to use when debatching messages.
* @param batchingStrategy the batching strategy.

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -143,7 +143,7 @@ public class EnableRabbitBatchIntegrationTests {
public SimpleRabbitListenerContainerFactory consumerBatchContainerFactory() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory());
factory.setBatchListener(true);
factory.setBatchListener(false);
factory.setConsumerBatchEnabled(true);
factory.setBatchSize(2);
return factory;
@@ -202,7 +202,7 @@ public class EnableRabbitBatchIntegrationTests {
this.fooMessagesLatch.countDown();
}
@RabbitListener(queues = "batch.3", containerFactory = "consumerBatchContainerFactory")
@RabbitListener(queues = "batch.3", containerFactory = "consumerBatchContainerFactory", batch = "true")
public void listen3(List<Foo> in) {
this.foosConsumerBatchToo = in;
this.fooConsumerBatchTooLatch.countDown();