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

GH-2490: Add forceStop to Container Factories

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

**cherry-pick to 2.4.x**
This commit is contained in:
Gary Russell
2023-07-19 13:38:43 -04:00
committed by GitHub
parent 58448ad5d3
commit fa944397e7
2 changed files with 19 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java
View File

@@ -117,6 +117,8 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
private RabbitListenerObservationConvention observationConvention;
private Boolean forceStop;
/**
* @param connectionFactory The connection factory.
* @see AbstractMessageListenerContainer#setConnectionFactory(ConnectionFactory)
@@ -339,6 +341,16 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
this.observationConvention = observationConvention;
}
/**
* Set to true to stop the container after the current message(s) are processed and
* requeue any prefetched. Useful when using exclusive or single-active consumers.
* @param forceStop true to stop when current messsage(s) are processed.
* @since 2.4.15
*/
public void setForceStop(boolean forceStop) {
this.forceStop = forceStop;
}
@Override
public C createListenerContainer(RabbitListenerEndpoint endpoint) {
C instance = createContainerInstance();
@@ -376,7 +388,8 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
.acceptIfNotNull(this.batchingStrategy, instance::setBatchingStrategy)
.acceptIfNotNull(getMicrometerEnabled(), instance::setMicrometerEnabled)
.acceptIfNotNull(getObservationEnabled(), instance::setObservationEnabled)
.acceptIfNotNull(this.observationConvention, instance::setObservationConvention);
.acceptIfNotNull(this.observationConvention, instance::setObservationConvention)
.acceptIfNotNull(this.forceStop, instance::setForceStop);
if (this.batchListener && this.deBatchingEnabled == null) {
// turn off container debatching by default for batch listeners
instance.setDeBatchingEnabled(false);

6
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/RabbitListenerContainerFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -114,6 +114,7 @@ public class RabbitListenerContainerFactoryTests {
this.factory.setAfterReceivePostProcessors(afterReceivePostProcessor);
this.factory.setGlobalQos(true);
this.factory.setContainerCustomizer(c -> c.setShutdownTimeout(10_000));
this.factory.setForceStop(true);
assertThat(this.factory.getAdviceChain()).isEqualTo(new Advice[]{advice});
@@ -150,6 +151,7 @@ public class RabbitListenerContainerFactoryTests {
assertThat(actualAfterReceivePostProcessors.size()).as("Wrong number of afterReceivePostProcessors").isEqualTo(1);
assertThat(actualAfterReceivePostProcessors.get(0)).as("Wrong advice").isSameAs(afterReceivePostProcessor);
assertThat(fieldAccessor.getPropertyValue("globalQos")).isEqualTo(true);
assertThat(TestUtils.getPropertyValue(container, "forceStop", Boolean.class)).isTrue();
}
@Test
@@ -176,6 +178,7 @@ public class RabbitListenerContainerFactoryTests {
this.direct.setMessagesPerAck(5);
this.direct.setAckTimeout(3L);
this.direct.setAfterReceivePostProcessors(afterReceivePostProcessor);
this.direct.setForceStop(true);
assertThat(this.direct.getAdviceChain()).isEqualTo(new Advice[]{advice});
@@ -207,6 +210,7 @@ public class RabbitListenerContainerFactoryTests {
List<?> actualAfterReceivePostProcessors = (List<?>) fieldAccessor.getPropertyValue("afterReceivePostProcessors");
assertThat(actualAfterReceivePostProcessors.size()).as("Wrong number of afterReceivePostProcessors").isEqualTo(1);
assertThat(actualAfterReceivePostProcessors.get(0)).as("Wrong afterReceivePostProcessor").isSameAs(afterReceivePostProcessor);
assertThat(TestUtils.getPropertyValue(container, "forceStop", Boolean.class)).isTrue();
}
private void setBasicConfig(AbstractRabbitListenerContainerFactory<?> factory) {