Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1201: Fix for ChannelAwareBatchMessageListener

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1201

The previous commit did not work with `ChannelAwareBatchMessageListener`.

Also add a test to get `List<o.s.a.c.Message>` in a `@RabbitListener` (which
exposed this issue).

**cherry-pick to 2.2.x**
This commit is contained in:
Gary Russell
2020-05-18 16:19:28 -04:00
committed by Artem Bilan
parent 751326ee63
commit 095b2df3c4
3 changed files with 45 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -56,6 +56,7 @@ import org.springframework.amqp.rabbit.connection.RabbitAccessor;
import org.springframework.amqp.rabbit.connection.RabbitResourceHolder;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.connection.RoutingConnectionFactory;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerExecutionException;
import org.springframework.amqp.rabbit.listener.exception.FatalListenerStartupException;
@@ -243,6 +244,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
private volatile boolean lazyLoad;
private boolean isBatchListener;
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
@@ -424,6 +427,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
*/
public void setMessageListener(MessageListener messageListener) {
this.messageListener = messageListener;
this.isBatchListener = messageListener instanceof BatchMessageListener
|| messageListener instanceof ChannelAwareBatchMessageListener;
}
/**
@@ -1923,8 +1928,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
@Nullable
protected List<Message> debatch(Message message) {
if (isDeBatchingEnabled() && getBatchingStrategy().canDebatch(message.getMessageProperties())
&& getMessageListener() instanceof BatchMessageListener) {
if (this.isBatchListener && isDeBatchingEnabled()
&& getBatchingStrategy().canDebatch(message.getMessageProperties())) {
final List<Message> messageList = new ArrayList<>();
getBatchingStrategy().deBatch(message, fragment -> messageList.add(fragment));
return messageList;

9
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -56,8 +56,10 @@ import org.springframework.amqp.rabbit.support.ConsumerCancelledException;
import org.springframework.amqp.rabbit.support.ListenerContainerAware;
import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.support.MetricType;
import org.springframework.lang.Nullable;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
@@ -819,8 +821,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
if (this.retryDeclarationInterval != null) {
consumer.setRetryDeclarationInterval(this.retryDeclarationInterval);
}
if (getConsumerTagStrategy() != null) {
consumer.setTagStrategy(getConsumerTagStrategy()); // NOSONAR never null here
ConsumerTagStrategy consumerTagStrategy = getConsumerTagStrategy();
if (consumerTagStrategy != null) {
consumer.setTagStrategy(consumerTagStrategy);
}
consumer.setBackOffExecution(getRecoveryBackOff().start());
consumer.setShutdownTimeout(getShutdownTimeout());
@@ -1084,7 +1087,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
@Override
protected void publishConsumerFailedEvent(String reason, boolean fatal, Throwable t) {
protected void publishConsumerFailedEvent(String reason, boolean fatal, @Nullable Throwable t) {
if (!fatal || !isRunning()) {
super.publishConsumerFailedEvent(reason, fatal, t);
}

34
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitBatchIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019 the original author or authors.
* Copyright 2019-2020 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.
@@ -34,6 +34,7 @@ import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -50,7 +51,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
*/
@SpringJUnitConfig
@DirtiesContext
@RabbitAvailable(queues = { "batch.1", "batch.2", "batch.3" })
@RabbitAvailable(queues = { "batch.1", "batch.2", "batch.3", "batch.4" })
public class EnableRabbitBatchIntegrationTests {
@Autowired
@@ -99,6 +100,20 @@ public class EnableRabbitBatchIntegrationTests {
assertThat(this.listener.foosConsumerBatchToo.get(3).getBar()).isEqualTo("qux");
}
@Test
public void nativeMessageList() throws InterruptedException {
this.template.convertAndSend("batch.4", new Foo("foo"));
this.template.convertAndSend("batch.4", new Foo("bar"));
assertThat(this.listener.nativeMessagesLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.nativeMessages).hasSize(2);
Foo payload = (Foo) new SimpleMessageConverter().fromMessage(this.listener.nativeMessages.get(0));
assertThat(payload.getBar()).isEqualTo("foo");
assertThat(this.listener.nativeMessages.get(1).getMessageProperties()
.getHeaders()
.get(AmqpHeaders.BATCH_SIZE))
.isEqualTo(2);
}
@Configuration
@EnableRabbit
public static class Config {
@@ -116,6 +131,11 @@ public class EnableRabbitBatchIntegrationTests {
DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory());
factory.setBatchListener(true);
factory.setContainerCustomizer(container -> {
if (container.getQueueNames()[0].equals("batch.4")) {
container.setDeBatchingEnabled(true);
}
});
return factory;
}
@@ -166,6 +186,10 @@ public class EnableRabbitBatchIntegrationTests {
CountDownLatch fooConsumerBatchTooLatch = new CountDownLatch(1);
private List<org.springframework.amqp.core.Message> nativeMessages;
private final CountDownLatch nativeMessagesLatch = new CountDownLatch(1);
@RabbitListener(queues = "batch.1")
public void listen1(List<Foo> in) {
this.foos = in;
@@ -184,6 +208,12 @@ public class EnableRabbitBatchIntegrationTests {
this.fooConsumerBatchTooLatch.countDown();
}
@RabbitListener(queues = "batch.4", containerFactory = "directListenerContainerFactory")
public void listen4(List<org.springframework.amqp.core.Message> in) {
this.nativeMessages = in;
this.nativeMessagesLatch.countDown();
}
}
@SuppressWarnings("serial")