From 07443ceb77272c165c12dc8154cac22da41dd10b Mon Sep 17 00:00:00 2001
From: zysaaa <982020642@qq.com>
Date: Tue, 14 Dec 2021 02:24:41 +0800
Subject: [PATCH] GH-1401: SMLC: Fix setConcurrency

Resolves https://github.com/spring-projects/spring-amqp/issues/1401

- multiple calls can temporarly inflate consumer count
---
 .../SimpleMessageListenerContainer.java       |  1 -
 .../SimpleMessageListenerContainerTests.java  | 28 ++++++++++++++++++-
 2 files changed, 27 insertions(+), 2 deletions(-)

diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
index b77ec1c1..338e9499 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
@@ -223,7 +223,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 				int maxConsumersToSet = Integer.parseInt(concurrency.substring(separatorIndex + 1));
 				Assert.isTrue(maxConsumersToSet >= consumersToSet,
 						"'maxConcurrentConsumers' value must be at least 'concurrentConsumers'");
-				this.concurrentConsumers = 1;
 				this.maxConcurrentConsumers = null;
 				setConcurrentConsumers(consumersToSet);
 				setMaxConcurrentConsumers(maxConsumersToSet);
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
index 32c133dd..ab880121 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -69,6 +69,7 @@ import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.core.AnonymousQueue;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageBuilder;
+import org.springframework.amqp.core.MessageListener;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
@@ -650,6 +651,31 @@ public class SimpleMessageListenerContainerTests {
 		assertThat(afterReceivePostProcessors).containsExactly(mpp2, mpp3);
 	}
 
+	@Test
+	void setConcurrency() throws Exception {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		Connection connection = mock(Connection.class);
+		Channel channel = mock(Channel.class);
+		given(connectionFactory.createConnection()).willReturn(connection);
+		given(connection.createChannel(false)).willReturn(channel);
+		final AtomicReference<Consumer> consumer = new AtomicReference<>();
+		willAnswer(invocation -> {
+			consumer.set(invocation.getArgument(6));
+			consumer.get().handleConsumeOk("1");
+			return "1";
+		}).given(channel)
+				.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(), anyMap(),
+						any(Consumer.class));
+		final SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		container.setQueueNames("foo", "bar");
+		container.setMessageListener(mock(MessageListener.class));
+		container.setConcurrency("5-10");
+		container.start();
+		await().until(() -> TestUtils.getPropertyValue(container, "consumers", Collection.class).size() == 5);
+		container.setConcurrency("10-10");
+		assertThat(TestUtils.getPropertyValue(container, "consumers", Collection.class)).hasSize(10);
+	}
+
 	private Answer<Object> messageToConsumer(final Channel mockChannel, final SimpleMessageListenerContainer container,
 			final boolean cancel, final CountDownLatch latch) {
 		return invocation -> {