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 fd0370fdd2..2645c21eda 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 @@ -54,6 +54,8 @@ public class SimplePool implements Pool { private final Set allocated = Collections.synchronizedSet(new HashSet()); + private final Set inUse = Collections.synchronizedSet(new HashSet()); + private final PoolItemCallback callback; /** @@ -127,7 +129,7 @@ public class SimplePool implements Pool { } public int getActiveCount() { - return this.getAllocatedCount() - this.getIdleCount(); + return this.inUse.size(); } public int getAllocatedCount() { @@ -192,32 +194,42 @@ public class SimplePool implements Pool { if (logger.isDebugEnabled()) { logger.debug("Received a stale item, will attempt to get a new one."); } + doRemoveItem(item); item = doGetItem(); } + this.inUse.add(item); return item; } /** - * Returns an item to the pool. Item may be null, in which case a subsequent getItem() - * will return a new instance. + * Returns an item to the pool. */ public synchronized void releaseItem(T item) { - Assert.isTrue(item == null || this.allocated.contains(item), + Assert.notNull(item, "Item cannot be null"); + Assert.isTrue(this.allocated.contains(item), "You can only release items that were obtained from the pool"); - if (this.poolSize.get() > targetPoolSize.get()) { - poolSize.decrementAndGet(); - if (item != null) { - doRemoveItem(item); + if (this.inUse.contains(item)) { + if (this.poolSize.get() > targetPoolSize.get()) { + poolSize.decrementAndGet(); + if (item != null) { + doRemoveItem(item); + } + } + else { + if (logger.isDebugEnabled()){ + logger.debug("Releasing " + item + " back to the pool"); + } + if (item != null) { + this.available.add(item); + this.inUse.remove(item); + } + permits.release(); } } else { if (logger.isDebugEnabled()){ - logger.debug("Releasing " + item + " back to the pool"); + logger.debug("Ignoring release of " + item + " back to the pool - not in use"); } - if (item != null) { - available.add(item); - } - permits.release(); } } @@ -230,6 +242,7 @@ public class SimplePool implements Pool { private void doRemoveItem(T item) { this.allocated.remove(item); + this.inUse.remove(item); this.callback.removedFromPool(item); } 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 0b2569de99..fc7a5cfb13 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-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -23,10 +23,13 @@ import static org.junit.Assert.fail; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; + import org.springframework.integration.MessagingException; +import org.springframework.integration.test.util.TestUtils; /** * @author Gary Russell @@ -51,6 +54,7 @@ public class SimplePoolTests { s3 = pool.getItem(); assertNotSame(s1, s3); assertFalse(strings.remove(s1)); + assertEquals(2, pool.getAllocatedCount()); } @Test @@ -128,6 +132,22 @@ public class SimplePoolTests { pool.releaseItem("Hello, world!"); } + @Test + public void testDoubleReturn() { + final Set strings = new HashSet(); + final AtomicBoolean stale = new AtomicBoolean(); + SimplePool pool = stringPool(2, strings, stale); + Semaphore permits = TestUtils.getPropertyValue(pool, "permits", Semaphore.class); + assertEquals(2, permits.availablePermits()); + String s1 = pool.getItem(); + assertEquals(1, permits.availablePermits()); + pool.releaseItem(s1); + assertEquals(2, permits.availablePermits()); + pool.releaseItem(s1); + assertEquals(2, permits.availablePermits()); + } + + private SimplePool stringPool(int size, final Set strings, final AtomicBoolean stale) { SimplePool pool = new SimplePool(size, new SimplePool.PoolItemCallback() { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java index 8535e90625..a88ef5f59c 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java @@ -16,29 +16,39 @@ package org.springframework.integration.file.remote.handler; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; import java.io.InputStream; +import java.util.Set; import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + import org.springframework.expression.ExpressionParser; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.Message; +import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.util.SimplePool; /** * @author Oleg Zhurakousky @@ -150,4 +160,40 @@ public class FileTransferringMessageHandlerTests { verify(session, times(0)).rename(Mockito.anyString(), Mockito.anyString()); } + @SuppressWarnings("unchecked") + @Test + public void testServerException() throws Exception{ + SessionFactory sf = mock(SessionFactory.class); + CachingSessionFactory csf = new CachingSessionFactory(sf, 2); + FileTransferringMessageHandler handler = new FileTransferringMessageHandler(csf); + Session session1 = newSession(); + Session session2 = newSession(); + Session session3 = newSession(); + when(sf.getSession()).thenReturn(session1, session2, session3); + handler.setRemoteDirectoryExpression(new LiteralExpression("foo")); + handler.afterPropertiesSet(); + for (int i = 0; i < 3; i++) { + try { + handler.handleMessage(new GenericMessage("hello")); + } + catch (Exception e) { + assertEquals("test", e.getCause().getCause().getMessage()); + } + } + verify(session1, times(1)).write(Mockito.any(InputStream.class), Mockito.anyString()); + verify(session2, times(1)).write(Mockito.any(InputStream.class), Mockito.anyString()); + verify(session3, times(1)).write(Mockito.any(InputStream.class), Mockito.anyString()); + SimplePool pool = TestUtils.getPropertyValue(csf, "pool", SimplePool.class); + assertEquals(1, pool.getAllocatedCount()); + assertEquals(1, pool.getIdleCount()); + assertSame(session3, TestUtils.getPropertyValue(pool, "allocated", Set.class).iterator().next()); + } + + private Session newSession() throws IOException { + @SuppressWarnings("unchecked") + Session session = mock(Session.class); + doThrow(new IOException("test")).when(session).write(any(InputStream.class), anyString()); + when(session.isOpen()).thenReturn(false); + return session; + } } 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 efe6117a6f..481088cbe6 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 @@ -103,23 +103,22 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public synchronized void close() { - /** - * If the delegate is stopped, actually close - * the connection. - */ - if (!isRunning()) { - if (logger.isDebugEnabled()){ - logger.debug("Factory not running - closing " + this.getConnectionId()); - } - pool.releaseItem(null); // just open up a permit - super.close(); - } - else if(this.released) { + if(this.released) { if (logger.isDebugEnabled()) { logger.debug("Connection " + this.getConnectionId() + " has already been released"); } } else { + /** + * If the delegate is stopped, actually close the connection, but still release + * it to the pool, it will be discarded/renewed the next time it is retrieved. + */ + if (!isRunning()) { + if (logger.isDebugEnabled()){ + logger.debug("Factory not running - closing " + this.getConnectionId()); + } + super.close(); + } pool.releaseItem(this.getTheConnection()); this.released = true; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java index 56dbe0168e..cc4ce38517 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java @@ -35,6 +35,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; @@ -124,6 +125,13 @@ public class CachingClientConnectionFactoryTests { when(factory.isRunning()).thenReturn(true); TcpConnectionSupport mockConn1 = makeMockConnection("conn1"); TcpConnectionSupport mockConn2 = makeMockConnection("conn2"); + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + return null; + } + }).when(mockConn1).close(); when(factory.getConnection()).thenReturn(mockConn1) .thenReturn(mockConn2).thenReturn(mockConn1) .thenReturn(mockConn2); @@ -197,6 +205,8 @@ public class CachingClientConnectionFactoryTests { conn2.close(); verify(mockConn1).close(); verify(mockConn2).close(); + when(mockConn1.isOpen()).thenReturn(false); + when(mockConn2.isOpen()).thenReturn(false); when(factory.isRunning()).thenReturn(true); TcpConnection conn3 = cachingFactory.getConnection(); assertNotSame(TestUtils.getPropertyValue(conn1, "theConnection"), @@ -209,8 +219,11 @@ public class CachingClientConnectionFactoryTests { public void testEnlargePool() throws Exception { AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class); when(factory.isRunning()).thenReturn(true); - TcpConnectionSupport mockConn = makeMockConnection("conn"); - when(factory.getConnection()).thenReturn(mockConn); + TcpConnectionSupport mockConn1 = makeMockConnection("conn1"); + TcpConnectionSupport mockConn2 = makeMockConnection("conn2"); + TcpConnectionSupport mockConn3 = makeMockConnection("conn3"); + TcpConnectionSupport mockConn4 = makeMockConnection("conn4"); + when(factory.getConnection()).thenReturn(mockConn1, mockConn2, mockConn3, mockConn4); CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(factory, 2); cachingFactory.start(); TcpConnection conn1 = cachingFactory.getConnection(); @@ -235,10 +248,10 @@ public class CachingClientConnectionFactoryTests { public void testReducePool() throws Exception { AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class); when(factory.isRunning()).thenReturn(true); - TcpConnectionSupport mockConn1 = makeMockConnection("conn", true); - TcpConnectionSupport mockConn2 = makeMockConnection("conn", true); - TcpConnectionSupport mockConn3 = makeMockConnection("conn", true); - TcpConnectionSupport mockConn4 = makeMockConnection("conn", true); + TcpConnectionSupport mockConn1 = makeMockConnection("conn1", true); + TcpConnectionSupport mockConn2 = makeMockConnection("conn2", true); + TcpConnectionSupport mockConn3 = makeMockConnection("conn3", true); + TcpConnectionSupport mockConn4 = makeMockConnection("conn4", true); when(factory.getConnection()).thenReturn(mockConn1) .thenReturn(mockConn2).thenReturn(mockConn3) .thenReturn(mockConn4);