From b56a56e8e630295dc05a388defd775559836236e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 18 Feb 2015 11:32:29 -0500 Subject: [PATCH] INT-3650: Fix TCP Caching Factory Configuration JIRA: https://jira.spring.io/browse/INT-3650 Don't propagate `setSingleUse` to the target factory. * Polishing for code style * Fix `CachingClientConnectionFactory#equals` --- .../CachingClientConnectionFactory.java | 211 +++++++++++------- .../CachingClientConnectionFactoryTests.java | 77 ++++++- 2 files changed, 199 insertions(+), 89 deletions(-) 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 8e451b19f1..780c3dc615 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp.connection; import java.util.concurrent.Executor; @@ -27,6 +28,11 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; /** + * Connection factory that caches connections from the underlying target factory. The underlying + * factory will be reconfigured to have {@code singleUse=true} in order for the connection to be + * returned to the cache after use. Users should not subsequently set the underlying property to + * false, or cache starvation will result. + * * @author Gary Russell * @since 2.2 * @@ -37,54 +43,87 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact private final SimplePool pool; + /** + * Construct a caching connection factory that delegates to the provided factory, with + * the provided pool size. + * @param target the target factory. + * @param poolSize the number of connections to allow. + */ public CachingClientConnectionFactory(AbstractClientConnectionFactory target, int poolSize) { super("", 0); // override single-use to true to force "close" after use target.setSingleUse(true); this.targetConnectionFactory = target; - pool = new SimplePool(poolSize, new SimplePool.PoolItemCallback() { + this.pool = new SimplePool(poolSize, + new SimplePool.PoolItemCallback() { - @Override - public TcpConnectionSupport createForPool() { - try { - return targetConnectionFactory.getConnection(); - } catch (Exception e) { - throw new MessagingException("Failed to obtain connection", e); - } - } + @Override + public TcpConnectionSupport createForPool() { + try { + return targetConnectionFactory.getConnection(); + } + catch (Exception e) { + throw new MessagingException("Failed to obtain connection", e); + } + } - @Override - public boolean isStale(TcpConnectionSupport connection) { - return !connection.isOpen(); - } + @Override + public boolean isStale(TcpConnectionSupport connection) { + return !connection.isOpen(); + } - @Override - public void removedFromPool(TcpConnectionSupport connection) { - connection.close(); - } - }); + @Override + public void removedFromPool(TcpConnectionSupport connection) { + connection.close(); + } + + }); } + /** + * @param connectionWaitTimeout the new timeout. + * @see SimplePool#setWaitTimeout(long) + */ public void setConnectionWaitTimeout(int connectionWaitTimeout) { this.pool.setWaitTimeout(connectionWaitTimeout); } + /** + * @param poolSize the new pool size. + * @see SimplePool#setPoolSize(int) + */ public synchronized void setPoolSize(int poolSize) { this.pool.setPoolSize(poolSize); } + /** + * @see SimplePool#getPoolSize() + * @return the pool size. + */ public int getPoolSize() { return this.pool.getPoolSize(); } + /** + * @see SimplePool#getIdleCount() + * @return the idle count. + */ public int getIdleCount() { return this.pool.getIdleCount(); } + /** + * @see SimplePool#getActiveCount() + * @return the active count. + */ public int getActiveCount() { return this.pool.getActiveCount(); } + /** + * @see SimplePool#getAllocatedCount() + * @return the allocated count. + */ public int getAllocatedCount() { return this.pool.getAllocatedCount(); } @@ -92,7 +131,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public TcpConnectionSupport obtainConnection() throws Exception { CachedConnection cachedConnection = new CachedConnection(this.pool.getItem()); - cachedConnection.registerListener(this.getListener()); + cachedConnection.registerListener(getListener()); return cachedConnection; } @@ -107,23 +146,23 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public synchronized void close() { - if(this.released) { + if (this.released) { if (logger.isDebugEnabled()) { - logger.debug("Connection " + this.getConnectionId() + " has already been released"); + logger.debug("Connection " + getConnectionId() + " has already been released"); } } - else { + 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()); + if (logger.isDebugEnabled()) { + logger.debug("Factory not running - closing " + getConnectionId()); } super.close(); } - pool.releaseItem(this.getTheConnection()); + pool.releaseItem(getTheConnection()); this.released = true; } } @@ -135,7 +174,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public String toString() { - return this.getConnectionId(); + return getConnectionId(); } /** @@ -146,20 +185,21 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact */ @Override public boolean onMessage(Message message) { - AbstractIntegrationMessageBuilder messageBuilder = CachingClientConnectionFactory.this - .getMessageBuilderFactory().fromMessage(message) - .setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId()); + AbstractIntegrationMessageBuilder messageBuilder = + CachingClientConnectionFactory.this.getMessageBuilderFactory() + .fromMessage(message) + .setHeader(IpHeaders.CONNECTION_ID, getConnectionId()); if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) { messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID)); } - this.getListener().onMessage(messageBuilder.build()); + getListener().onMessage(messageBuilder.build()); close(); // return to pool after response is received return true; // true so the single-use connection doesn't close itself } private void physicallyClose() { - this.getTheConnection().close(); + getTheConnection().close(); } } @@ -168,127 +208,133 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public boolean isRunning() { - return targetConnectionFactory.isRunning(); + return this.targetConnectionFactory.isRunning(); } @Override public int hashCode() { - return targetConnectionFactory.hashCode(); + return this.targetConnectionFactory.hashCode(); } @Override public void setComponentName(String componentName) { - targetConnectionFactory.setComponentName(componentName); + this.targetConnectionFactory.setComponentName(componentName); } @Override public String getComponentType() { - return targetConnectionFactory.getComponentType(); + return this.targetConnectionFactory.getComponentType(); } @Override - public boolean equals(Object obj) { - return targetConnectionFactory.equals(obj); + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + CachingClientConnectionFactory that = (CachingClientConnectionFactory) o; + + return this.targetConnectionFactory.equals(that.targetConnectionFactory); + } @Override public int getSoTimeout() { - return targetConnectionFactory.getSoTimeout(); + return this.targetConnectionFactory.getSoTimeout(); } @Override public void setSoTimeout(int soTimeout) { - targetConnectionFactory.setSoTimeout(soTimeout); + this.targetConnectionFactory.setSoTimeout(soTimeout); } @Override public int getSoReceiveBufferSize() { - return targetConnectionFactory.getSoReceiveBufferSize(); + return this.targetConnectionFactory.getSoReceiveBufferSize(); } @Override public void setSoReceiveBufferSize(int soReceiveBufferSize) { - targetConnectionFactory.setSoReceiveBufferSize(soReceiveBufferSize); + this.targetConnectionFactory.setSoReceiveBufferSize(soReceiveBufferSize); } @Override public int getSoSendBufferSize() { - return targetConnectionFactory.getSoSendBufferSize(); + return this.targetConnectionFactory.getSoSendBufferSize(); } @Override public void setSoSendBufferSize(int soSendBufferSize) { - targetConnectionFactory.setSoSendBufferSize(soSendBufferSize); + this.targetConnectionFactory.setSoSendBufferSize(soSendBufferSize); } @Override public boolean isSoTcpNoDelay() { - return targetConnectionFactory.isSoTcpNoDelay(); + return this.targetConnectionFactory.isSoTcpNoDelay(); } @Override public void setSoTcpNoDelay(boolean soTcpNoDelay) { - targetConnectionFactory.setSoTcpNoDelay(soTcpNoDelay); + this.targetConnectionFactory.setSoTcpNoDelay(soTcpNoDelay); } @Override public int getSoLinger() { - return targetConnectionFactory.getSoLinger(); + return this.targetConnectionFactory.getSoLinger(); } @Override public void setSoLinger(int soLinger) { - targetConnectionFactory.setSoLinger(soLinger); + this.targetConnectionFactory.setSoLinger(soLinger); } @Override public boolean isSoKeepAlive() { - return targetConnectionFactory.isSoKeepAlive(); + return this.targetConnectionFactory.isSoKeepAlive(); } @Override public void setSoKeepAlive(boolean soKeepAlive) { - targetConnectionFactory.setSoKeepAlive(soKeepAlive); + this.targetConnectionFactory.setSoKeepAlive(soKeepAlive); } @Override public int getSoTrafficClass() { - return targetConnectionFactory.getSoTrafficClass(); + return this.targetConnectionFactory.getSoTrafficClass(); } @Override public void setSoTrafficClass(int soTrafficClass) { - targetConnectionFactory.setSoTrafficClass(soTrafficClass); + this.targetConnectionFactory.setSoTrafficClass(soTrafficClass); } @Override public String getHost() { - return targetConnectionFactory.getHost(); + return this.targetConnectionFactory.getHost(); } @Override public int getPort() { - return targetConnectionFactory.getPort(); + return this.targetConnectionFactory.getPort(); } @Override public TcpSender getSender() { - return targetConnectionFactory.getSender(); + return this.targetConnectionFactory.getSender(); } @Override public Serializer getSerializer() { - return targetConnectionFactory.getSerializer(); + return this.targetConnectionFactory.getSerializer(); } @Override public Deserializer getDeserializer() { - return targetConnectionFactory.getDeserializer(); + return this.targetConnectionFactory.getDeserializer(); } @Override public TcpMessageMapper getMapper() { - return targetConnectionFactory.getMapper(); + return this.targetConnectionFactory.getMapper(); } /** @@ -315,7 +361,8 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public void registerListener(TcpListener listener) { super.registerListener(listener); - targetConnectionFactory.registerListener(new TcpListener() { + this.targetConnectionFactory.registerListener(new TcpListener() { + @Override public boolean onMessage(Message message) { if (!(message instanceof ErrorMessage)) { @@ -323,59 +370,67 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact } return false; } + }); } @Override public void registerSender(TcpSender sender) { - targetConnectionFactory.registerSender(sender); + this.targetConnectionFactory.registerSender(sender); } @Override public void setTaskExecutor(Executor taskExecutor) { - targetConnectionFactory.setTaskExecutor(taskExecutor); + this.targetConnectionFactory.setTaskExecutor(taskExecutor); } @Override public void setDeserializer(Deserializer deserializer) { - targetConnectionFactory.setDeserializer(deserializer); + this.targetConnectionFactory.setDeserializer(deserializer); } @Override public void setSerializer(Serializer serializer) { - targetConnectionFactory.setSerializer(serializer); + this.targetConnectionFactory.setSerializer(serializer); } @Override public void setMapper(TcpMessageMapper mapper) { - targetConnectionFactory.setMapper(mapper); + this.targetConnectionFactory.setMapper(mapper); } @Override public boolean isSingleUse() { - return targetConnectionFactory.isSingleUse(); + return this.targetConnectionFactory.isSingleUse(); } + /** + * Ignored on this factory; connections are always cached in the pool. The underlying + * connection factory will have its singleUse property coerced to true (causing the + * connection to be returned). Setting it to false on the underlying factory after initialization + * will cause cache starvation. + * @param singleUse the singleUse. + */ @Override public void setSingleUse(boolean singleUse) { - targetConnectionFactory.setSingleUse(singleUse); + if (!singleUse && logger.isDebugEnabled()) { + logger.debug("singleUse=false is not supported; cached connections are never closed"); + } } @Override - public void setInterceptorFactoryChain( - TcpConnectionInterceptorFactoryChain interceptorFactoryChain) { - targetConnectionFactory - .setInterceptorFactoryChain(interceptorFactoryChain); + public void setInterceptorFactoryChain(TcpConnectionInterceptorFactoryChain interceptorFactoryChain) { + this.targetConnectionFactory.setInterceptorFactoryChain(interceptorFactoryChain); } @Override public void setLookupHost(boolean lookupHost) { - targetConnectionFactory.setLookupHost(lookupHost); + this.targetConnectionFactory.setLookupHost(lookupHost); } @Override public boolean isLookupHost() { - return targetConnectionFactory.isLookupHost(); + return this.targetConnectionFactory.isLookupHost(); } @@ -390,30 +445,30 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public void start() { - this.setActive(true); - targetConnectionFactory.start(); + setActive(true); + this.targetConnectionFactory.start(); super.start(); } @Override public synchronized void stop() { - targetConnectionFactory.stop(); + this.targetConnectionFactory.stop(); this.pool.removeAllIdleItems(); } @Override public int getPhase() { - return targetConnectionFactory.getPhase(); + return this.targetConnectionFactory.getPhase(); } @Override public boolean isAutoStartup() { - return targetConnectionFactory.isAutoStartup(); + return this.targetConnectionFactory.isAutoStartup(); } @Override public void stop(Runnable callback) { - targetConnectionFactory.stop(callback); + this.targetConnectionFactory.stop(callback); } } 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 a3b11056a2..9e1cd97ba4 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 @@ -38,8 +38,10 @@ import java.nio.ByteBuffer; import java.nio.channels.SocketChannel; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; @@ -66,6 +68,7 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.DirtiesContext.ClassMode; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.SocketUtils; /** * @author Gary Russell @@ -74,7 +77,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) -@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD) +@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD) public class CachingClientConnectionFactoryTests { @Autowired @@ -147,6 +150,7 @@ public class CachingClientConnectionFactoryTests { public Object answer(InvocationOnMock invocation) throws Throwable { return null; } + }).when(mockConn1).close(); when(factory.getConnection()).thenReturn(mockConn1) .thenReturn(mockConn2).thenReturn(mockConn1) @@ -166,11 +170,11 @@ public class CachingClientConnectionFactoryTests { TcpConnection conn2a = cachingFactory.getConnection(); assertEquals("Cached:" + mockConn2.toString(), conn2a.toString()); assertSame(TestUtils.getPropertyValue(conn2, "theConnection"), - TestUtils.getPropertyValue(conn2a, "theConnection")); + TestUtils.getPropertyValue(conn2a, "theConnection")); conn2a.close(); } - @Test(expected=MessagingException.class) + @Test(expected = MessagingException.class) public void testLimit() throws Exception { AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class); when(factory.isRunning()).thenReturn(true); @@ -210,11 +214,14 @@ public class CachingClientConnectionFactoryTests { TcpConnection conn2 = cachingFactory.getConnection(); assertEquals("Cached:" + mockConn2.toString(), conn2.toString()); cachingFactory.stop(); - Answer answer = new Answer () { + Answer answer = new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; - }}; + } + + }; doAnswer(answer).when(mockConn1).close(); doAnswer(answer).when(mockConn2).close(); when(factory.isRunning()).thenReturn(false); @@ -227,9 +234,9 @@ public class CachingClientConnectionFactoryTests { when(factory.isRunning()).thenReturn(true); TcpConnection conn3 = cachingFactory.getConnection(); assertNotSame(TestUtils.getPropertyValue(conn1, "theConnection"), - TestUtils.getPropertyValue(conn3, "theConnection")); + TestUtils.getPropertyValue(conn3, "theConnection")); assertNotSame(TestUtils.getPropertyValue(conn2, "theConnection"), - TestUtils.getPropertyValue(conn3, "theConnection")); + TestUtils.getPropertyValue(conn3, "theConnection")); } @Test @@ -359,9 +366,9 @@ public class CachingClientConnectionFactoryTests { @Override public void publishEvent(Object event) { - + } - + }, "foo"); conn.setMapper(new TcpMessageMapper()); conn.setSerializer(new ByteArrayCrLfSerializer()); @@ -381,9 +388,9 @@ public class CachingClientConnectionFactoryTests { @Override public void publishEvent(Object event) { - + } - + }, "foo"); conn.setMapper(new TcpMessageMapper()); conn.setSerializer(new ByteArrayCrLfSerializer()); @@ -425,6 +432,7 @@ public class CachingClientConnectionFactoryTests { final List connectionIds = new ArrayList(); final AtomicBoolean okToRun = new AtomicBoolean(true); Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override public void run() { while (okToRun.get()) { @@ -437,6 +445,7 @@ public class CachingClientConnectionFactoryTests { } } } + }); TestingUtilities.waitListening(serverCf, null); toGateway.send(new GenericMessage("Hello, world!")); @@ -501,10 +510,12 @@ public class CachingClientConnectionFactoryTests { when(factory2.isActive()).thenReturn(true); doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class)); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { return null; } + }).when(mockConn2).send(Mockito.any(Message.class)); FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories); failoverFactory.start(); @@ -519,6 +530,50 @@ public class CachingClientConnectionFactoryTests { Mockito.verify(mockConn2).send(message); } + @Test //INT-3650 + public void testRealConnection() throws Exception { + int port = SocketUtils.findAvailableTcpPort(); + TcpNetServerConnectionFactory in = new TcpNetServerConnectionFactory(port); + final CountDownLatch latch1 = new CountDownLatch(2); + final CountDownLatch latch2 = new CountDownLatch(102); + final List connectionIds = new ArrayList(); + in.registerListener(new TcpListener() { + + @Override + public boolean onMessage(Message message) { + connectionIds.add((String) message.getHeaders().get(IpHeaders.CONNECTION_ID)); + latch1.countDown(); + latch2.countDown(); + return false; + } + + }); + in.start(); + int n = 0; + while (n++ < 100 && !in.isListening()) { + Thread.sleep(100); + } + assertTrue(in.isListening()); + TcpNetClientConnectionFactory out = new TcpNetClientConnectionFactory("localhost", port); + CachingClientConnectionFactory cache = new CachingClientConnectionFactory(out, 1); + cache.setSingleUse(false); + cache.start(); + TcpConnectionSupport connection1 = cache.getConnection(); + connection1.send(new GenericMessage("foo")); + TcpConnectionSupport connection2 = cache.getConnection(); + connection2.send(new GenericMessage("foo")); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertSame(connectionIds.get(0), connectionIds.get(1)); + for (int i = 0; i < 100; i++) { + TcpConnectionSupport connection = cache.getConnection(); + connection.send(new GenericMessage("foo")); + } + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertSame(connectionIds.get(0), connectionIds.get(101)); + in.stop(); + cache.stop(); + } + public TcpConnectionSupport makeMockConnection() { TcpConnectionSupport connection = mock(TcpConnectionSupport.class); when(connection.isOpen()).thenReturn(true);