GH-3143: Fix simple pool resizing

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

Resize down does not work when not enough already allocated to match the reduction.

Improved SimplePool test cases and fixed code style

Fixed code style

GH-3143: Test Polishing

- use preferred assertJ expected exception checking
- add a test for reducing a partially allocated pool
- check active counts

spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java

@@ -36,6 +36,7 @@ import org.springframework.util.Assert;
  * demand up to the limit.
  *
  * @author Gary Russell
+ * @author Sergey Bogatyrev
  * @since 2.2
  *
  */
@@ -102,11 +103,9 @@ public class SimplePool<T> implements Pool<T> {
 					break;
 				}
 				T item = this.available.poll();
-				if (item == null) {
-					this.permits.release();
-					break;
+				if (item != null) {
+					doRemoveItem(item);
 				}
-				doRemoveItem(item);
 				this.poolSize.decrementAndGet();
 				delta++;
 			}

spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java

@@ -17,9 +17,12 @@
 package org.springframework.integration.util;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.fail;
 
+import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -30,6 +33,7 @@ import org.springframework.integration.test.util.TestUtils;
 
 /**
  * @author Gary Russell
+ * @author Sergey Bogatyrev
  * @since 2.2
  *
  */
@@ -147,6 +151,105 @@ public class SimplePoolTests {
 		assertThat(permits.availablePermits()).isEqualTo(2);
 	}
 
+	@Test
+	public void testSizeUpdateIfNotAllocated() {
+		SimplePool<String> pool = stringPool(10, new HashSet<>(), new AtomicBoolean());
+		pool.setWaitTimeout(0);
+		pool.setPoolSize(5);
+		assertThat(pool.getPoolSize()).isEqualTo(5);
+
+		// allocating all available items to check permits
+		Set<String> allocatedItems = new HashSet<>();
+		for (int i = 0; i < 5; i++) {
+			allocatedItems.add(pool.getItem());
+		}
+		assertThat(allocatedItems).hasSize(5);
+
+		// no more items can be allocated (indirect check of permits)
+		assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem());
+	}
+
+	@Test
+	public void testSizeUpdateIfPartiallyAllocated() {
+		SimplePool<String> pool = stringPool(10, new HashSet<>(), new AtomicBoolean());
+		pool.setWaitTimeout(0);
+		List<String> allocated = new ArrayList<>();
+		for (int i = 0; i < 10; i++) {
+			allocated.add(pool.getItem());
+		}
+
+		// release only 2 items
+		for (int i = 0; i < 2; i++) {
+			pool.releaseItem(allocated.get(i));
+		}
+
+		// trying to reduce pool size
+		pool.setPoolSize(5);
+
+		// at this moment the actual pool size can be reduced only partially, because
+		// only 2 items have been released, so 8 items are in use
+		assertThat(pool.getPoolSize()).isEqualTo(8);
+		assertThat(pool.getAllocatedCount()).isEqualTo(8);
+		assertThat(pool.getIdleCount()).isEqualTo(0);
+		assertThat(pool.getActiveCount()).isEqualTo(8);
+
+		// releasing 3 items
+		for (int i = 2; i < 5; i++) {
+			pool.releaseItem(allocated.get(i));
+		}
+
+		// now pool size should be reduced
+		assertThat(pool.getPoolSize()).isEqualTo(5);
+		assertThat(pool.getAllocatedCount()).isEqualTo(5);
+		assertThat(pool.getIdleCount()).isEqualTo(0);
+		assertThat(pool.getActiveCount()).isEqualTo(5);
+
+		// no more items can be allocated (indirect check of permits)
+		assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem());
+	}
+
+	@Test
+	public void testSizeUpdateIfFullyAllocated() {
+		SimplePool<String> pool = stringPool(10, new HashSet<>(), new AtomicBoolean());
+		pool.setWaitTimeout(0);
+		List<String> allocated = new ArrayList<>();
+		for (int i = 0; i < 10; i++) {
+			allocated.add(pool.getItem());
+		}
+
+		// trying to reduce pool size
+		pool.setPoolSize(5);
+
+		// at this moment the actual pool size cannot be reduced - all in use
+		assertThat(pool.getPoolSize()).isEqualTo(10);
+		assertThat(pool.getAllocatedCount()).isEqualTo(10);
+		assertThat(pool.getIdleCount()).isEqualTo(0);
+		assertThat(pool.getActiveCount()).isEqualTo(10);
+
+		// releasing 5 items
+		for (int i = 0; i < 5; i++) {
+			pool.releaseItem(allocated.get(i));
+		}
+
+		// now pool size should be reduced
+		assertThat(pool.getPoolSize()).isEqualTo(5);
+		assertThat(pool.getAllocatedCount()).isEqualTo(5);
+		assertThat(pool.getIdleCount()).isEqualTo(0);
+		assertThat(pool.getActiveCount()).isEqualTo(5);
+
+		// no more items can be allocated (indirect check of permits)
+		assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem());
+
+		// releasing remaining items
+		for (int i = 5; i < 10; i++) {
+			pool.releaseItem(allocated.get(i));
+		}
+
+		assertThat(pool.getPoolSize()).isEqualTo(5);
+		assertThat(pool.getAllocatedCount()).isEqualTo(5);
+		assertThat(pool.getIdleCount()).isEqualTo(5);
+		assertThat(pool.getActiveCount()).isEqualTo(0);
+	}
 
 	private SimplePool<String> stringPool(int size, final Set<String> strings,
 			final AtomicBoolean stale) {