From 32251bf6f2ff3e1c9bb6cb8ac3be7b4838755f01 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 29 Nov 2012 14:59:20 -0500 Subject: [PATCH] INT-2839 Fix TCP Caching Connections with Gateways Using the CachingClientConnectionFactory with a gateway doesn't work. The Listener (gateway) was not being set up properly by the wrapper. Instead, the actualListener in the wrapped connection was set to null. This caused the connection to "close" itself (return to the pool) immediately after the send. In addition, even with the actualListener set up to properly reference the gateway, it still wouldn't work because the gateway can't correlate the reply (the cached connectionId is prefixed with "cached:" so the gateway can't find the reply. Further, when onMessage() returned, it caused the reader loop to end. Fixes: 1. Properly set up the cached connection with the real listener so that ultimately the underlying connection sees there is an actualListener and so doesn't close itself after the send. 2. Override getListener() on the cached connection to return the real listener. 3. Override onMessage() in order to fix up the connection id in the message, and return true (intercepted) so the connection doesn't close itself (terminate the reader thread). 4. close() (return to the pool) after invoking the gateway's onMessage(). 5. Add test cases to verify proper operation including an assertion that the pooled connection is reused. This is made a little more complex by the tangle between connections and connection interceptors; mainly because single-use connections close themselves after use. This will be addressed in 3.0 (INT-2829) making connection interceptors simpler. --- .../CachingClientConnectionFactory.java | 27 ++++++++++ ...ngClientConnectionFactoryTests-context.xml | 31 ++++++++++++ .../CachingClientConnectionFactoryTests.java | 50 +++++++++++++++++++ 3 files changed, 108 insertions(+) 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 744c29d393..0369a12903 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,7 +21,10 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; +import org.springframework.integration.Message; import org.springframework.integration.MessagingException; +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.util.SimplePool; /** @@ -37,6 +40,8 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact private final SimplePool pool; + private volatile TcpListener listener; + public CachingClientConnectionFactory(AbstractClientConnectionFactory target, int poolSize) { super("", 0); // override single-use to true to force "close" after use @@ -140,6 +145,27 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact return this.getConnectionId(); } + @Override + public TcpListener getListener() { + return CachingClientConnectionFactory.this.listener; + } + + /** + * We have to intercept the message to replace the connectionId header with + * ours so the listener can correlate a response with a request. We supply + * the actual connectionId in another header for convenience and tracing + * purposes. + */ + @Override + public boolean onMessage(Message message) { + CachingClientConnectionFactory.this.listener.onMessage(MessageBuilder.fromMessage(message) + .setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId()) + .setHeader(IpHeaders.ACTUAL_CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID)) + .build()); + close(); // return to pool after response is received + return true; // true so the single-use connection doesn't close itself + } + } ///////////////// DELEGATE METHODS /////////////////////// @@ -281,6 +307,7 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact @Override public void registerListener(TcpListener listener) { + this.listener = listener; targetConnectionFactory.registerListener(listener); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml index 6562547d03..ce3986f640 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests-context.xml @@ -22,6 +22,12 @@ + + + + + + + + + + + + + + + + + + + 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 b9905e15b7..2601ad8276 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 @@ -25,7 +25,11 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import org.junit.runner.RunWith; @@ -39,6 +43,7 @@ import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -61,6 +66,15 @@ public class CachingClientConnectionFactoryTests { @Autowired AbstractServerConnectionFactory serverCf; + @Autowired + SubscribableChannel toGateway; + + @Autowired + SubscribableChannel replies; + + @Autowired + PollableChannel fromGateway; + @Test public void testReuse() throws Exception { AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class); @@ -275,10 +289,46 @@ public class CachingClientConnectionFactoryTests { Message m = inbound.receive(1000); assertNotNull(m); String connectionId = m.getHeaders().get(IpHeaders.CONNECTION_ID, String.class); + + // assert we use the same connection from the pool outbound.send(new GenericMessage("Hello, world!")); m = inbound.receive(1000); assertNotNull(m); assertEquals(connectionId, m.getHeaders().get(IpHeaders.CONNECTION_ID, String.class)); + } + @Test + public void gatewayIntegrationTest() throws Exception { + final List connectionIds = new ArrayList(); + final AtomicBoolean okToRun = new AtomicBoolean(true); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + while (okToRun.get()) { + Message m = inbound.receive(1000); + if (m != null) { + connectionIds.add((String) m.getHeaders().get(IpHeaders.CONNECTION_ID)); + replies.send(MessageBuilder.withPayload("foo:" + new String((byte[]) m.getPayload())) + .copyHeaders(m.getHeaders()) + .build()); + } + } + } + }); + TestingUtilities.waitListening(serverCf, null); + toGateway.send(new GenericMessage("Hello, world!")); + Message m = fromGateway.receive(1000); + assertNotNull(m); + assertEquals("foo:" + "Hello, world!", new String((byte[]) m.getPayload())); + + // assert we use the same connection from the pool + toGateway.send(new GenericMessage("Hello, world2!")); + m = fromGateway.receive(1000); + assertNotNull(m); + assertEquals("foo:" + "Hello, world2!", new String((byte[]) m.getPayload())); + + assertEquals(2, connectionIds.size()); + assertEquals(connectionIds.get(0), connectionIds.get(1)); + + okToRun.set(false); } }