diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java index 8a201d0214..351ae22f67 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/Pool.java @@ -72,4 +72,11 @@ public interface Pool { */ int getAllocatedCount(); + /** + * Close the pool; returned items will be destroyed. + * @since 4.3.23 + */ + default void close() { + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index 5178761fe9..8cb5d46b32 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -62,6 +62,8 @@ public class SimplePool implements Pool { private final PoolItemCallback callback; + private volatile boolean closed; + /** * Creates a SimplePool with a specific limit. * @param poolSize The maximum number of items the pool supports. @@ -161,6 +163,7 @@ public class SimplePool implements Pool { */ @Override public T getItem() { + Assert.state(!this.closed, "Pool has been closed"); boolean permitted = false; try { try { @@ -218,7 +221,7 @@ public class SimplePool implements Pool { Assert.isTrue(this.allocated.contains(item), "You can only release items that were obtained from the pool"); if (this.inUse.contains(item)) { - if (this.poolSize.get() > this.targetPoolSize.get()) { + if (this.poolSize.get() > this.targetPoolSize.get() || this.closed) { this.poolSize.decrementAndGet(); doRemoveItem(item); } @@ -255,6 +258,12 @@ public class SimplePool implements Pool { this.callback.removedFromPool(item); } + @Override + public synchronized void close() { + this.closed = true; + removeAllIdleItems(); + } + /** * User of the pool provide an implementation of this interface; called during * various pool operations. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java index aea5015642..883390f568 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-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. @@ -18,6 +18,8 @@ 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.assertThatIllegalArgumentException; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import static org.assertj.core.api.Assertions.fail; import java.util.ArrayList; @@ -27,7 +29,7 @@ import java.util.Set; import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicBoolean; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.integration.test.util.TestUtils; @@ -127,13 +129,13 @@ public class SimplePoolTests { assertThat(pool.getAllocatedCount()).isEqualTo(2); } - @Test(expected = IllegalArgumentException.class) + @Test public void testForeignObject() { final Set strings = new HashSet(); final AtomicBoolean stale = new AtomicBoolean(); SimplePool pool = stringPool(2, strings, stale); pool.getItem(); - pool.releaseItem("Hello, world!"); + assertThatIllegalArgumentException().isThrownBy(() -> pool.releaseItem("Hello, world!")); } @Test @@ -251,16 +253,32 @@ public class SimplePoolTests { assertThat(pool.getActiveCount()).isEqualTo(0); } + @Test + void testClose() { + SimplePool pool = stringPool(10, new HashSet<>(), new AtomicBoolean()); + String item1 = pool.getItem(); + String item2 = pool.getItem(); + pool.releaseItem(item2); + assertThat(pool.getAllocatedCount()).isEqualTo(2); + pool.close(); + pool.releaseItem(item1); + assertThat(pool.getAllocatedCount()).isEqualTo(0); + assertThatIllegalStateException().isThrownBy(pool::getItem); + } + private SimplePool stringPool(int size, final Set strings, final AtomicBoolean stale) { + SimplePool pool = new SimplePool(size, new SimplePool.PoolItemCallback() { private int i; + @Override public String createForPool() { String string = "String" + i++; strings.add(string); return string; } + @Override public boolean isStale(String item) { if (stale.get()) { @@ -268,10 +286,12 @@ public class SimplePoolTests { } return stale.get(); } + @Override public void removedFromPool(String item) { strings.remove(item); } + }); return pool; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java index e5c8cee7f7..5a16632a04 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java @@ -137,7 +137,7 @@ public class CachingSessionFactory implements SessionFactory, DisposableBe */ @Override public void destroy() { - this.pool.removeAllIdleItems(); + this.pool.close(); } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java index 354ae842de..afb34e68ce 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java @@ -21,6 +21,7 @@ import java.util.Map; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicBoolean; +import org.springframework.beans.factory.DisposableBean; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.integration.ip.IpHeaders; @@ -41,7 +42,7 @@ import org.springframework.messaging.support.ErrorMessage; * @since 2.2 * */ -public class CachingClientConnectionFactory extends AbstractClientConnectionFactory { +public class CachingClientConnectionFactory extends AbstractClientConnectionFactory implements DisposableBean { private final AbstractClientConnectionFactory targetConnectionFactory; @@ -389,6 +390,11 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact this.pool.removeAllIdleItems(); } + @Override + public void destroy() throws Exception { + this.pool.close(); + } + private final class CachedConnection extends TcpConnectionInterceptorSupport { private final AtomicBoolean released = new AtomicBoolean(); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java index d04d65ec48..7425d2814c 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -30,6 +30,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.channel.QueueChannel; @@ -134,6 +135,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { .isIn(" sftpSource1.txt", "sftpSource2.txt"); received.getPayload().close(); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); } @Test @@ -148,6 +150,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { .isIn(" sftpSource1.txt", "sftpSource2.txt"); received.getPayload().close(); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); } @Test @@ -162,6 +165,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport { .isIn(" sftpSource1.txt", "sftpSource2.txt"); received.getPayload().close(); + StaticMessageHeaderAccessor.getCloseableResource(received).close(); } private SftpStreamingMessageSource buildSource() {