Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3096 TCP Allow Stacking Cache and Failover

Browse Source 
Support nesting the fail over and caching client connection
factories.

The TcpListener chain was not set up properly; this prevented
caching Failover connections, or Failover cached connections.

Each nested interceptor needs the next outer interceptor set
up as its listener, so the underlying connection calls
onMessage() all the way up to the ultimate listener (adapter
or gateway).

Fix the Listener hierarchy; add tests for caching failover
connections and failing over cached connections.

Don't overrwrite the actual connection id if it has already
been set up by a lower level wrapper.
This commit is contained in:
Gary Russell
2013-08-18 13:16:43 -04:00
parent e73a5dc69f
commit 351af6b16b
4 changed files with 265 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java
View File

@@ -36,8 +36,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
private final SimplePool<TcpConnectionSupport> pool;
private volatile TcpListener listener;
public CachingClientConnectionFactory(AbstractClientConnectionFactory target, int poolSize) {
super("", 0);
// override single-use to true to force "close" after use
@@ -89,7 +87,9 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
@Override
public TcpConnectionSupport obtainConnection() throws Exception {
return new CachedConnection(this.pool.getItem());
CachedConnection cachedConnection = new CachedConnection(this.pool.getItem());
cachedConnection.registerListener(this.getListener());
return cachedConnection;
}
private class CachedConnection extends TcpConnectionInterceptorSupport {
@@ -134,11 +134,6 @@ 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
@@ -147,10 +142,13 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
*/
@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());
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.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());
close(); // return to pool after response is received
return true; // true so the single-use connection doesn't close itself
}
@@ -273,11 +271,6 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
return targetConnectionFactory.getPort();
}
@Override
public TcpListener getListener() {
return targetConnectionFactory.getListener();
}
@Override
public TcpSender getSender() {
return targetConnectionFactory.getSender();
@@ -298,10 +291,35 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
return targetConnectionFactory.getMapper();
}
/**
* Delegate TCP Client Connection factories that are used to receive
* data need a Listener to send the messages to.
* This applies to client factories used for outbound gateways
* or for a pair of collaborating channel adapters.
* <p>
* During initialization, if a factory detects it has no listener
* it's listening logic (active thread) is terminated.
* <p>
* The listener registered with a factory is provided to each
* connection it creates so it can call the onMessage() method.
* <p>
* This code satisfies the first requirement in that this
* listener signals to the factory that it needs to run
* its listening logic.
* <p>
* When we wrap actual connections with CachedConnections,
* the connection is given the wrapper as a listener, so it
* can enhance the headers in onMessage(); the wrapper then invokes
* the real listener supplied here, with the modified message.
*/
@Override
public void registerListener(TcpListener listener) {
this.listener = listener;
targetConnectionFactory.registerListener(listener);
super.registerListener(listener);
targetConnectionFactory.registerListener(new TcpListener() {
public boolean onMessage(Message<?> message) {
throw new UnsupportedOperationException("This should never be called");
}
});
}
@Override

25
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java
View File

@@ -100,7 +100,9 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
if (connection != null && connection.isOpen()) {
return connection;
}
return new FailoverTcpConnection(this.factories);
FailoverTcpConnection failoverTcpConnection = new FailoverTcpConnection(this.factories);
failoverTcpConnection.registerListener(this.getListener());
return failoverTcpConnection;
}
@Override
@@ -279,11 +281,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
return this.delegate.getDeserializerStateKey();
}
@Override
public void registerListener(TcpListener listener) {
this.delegate.registerListener(listener);
}
@Override
public void registerSender(TcpSender sender) {
this.delegate.registerSender(sender);
@@ -334,11 +331,6 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
this.delegate.setSerializer(serializer);
}
@Override
public TcpListener getListener() {
return this.delegate.getListener();
}
@Override
public long incrementAndGetConnectionSequence() {
return this.delegate.incrementAndGetConnectionSequence();
@@ -351,10 +343,13 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
* purposes.
*/
public boolean onMessage(Message<?> message) {
return FailoverClientConnectionFactory.this.getListener().onMessage(MessageBuilder.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId())
.setHeader(IpHeaders.ACTUAL_CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID))
.build());
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId());
if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));
}
return this.getListener().onMessage(messageBuilder.build());
}
}

176
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

@@ -20,6 +20,10 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.ObjectInputStream;
@@ -45,16 +49,21 @@ import javax.net.ServerSocketFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.core.serializer.DefaultDeserializer;
import org.springframework.core.serializer.DefaultSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.FailoverClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnectionSupport;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
import org.springframework.integration.test.util.TestUtils;
@@ -178,6 +187,7 @@ public class TcpOutboundGatewayTests {
assertTrue(replies.remove("Reply" + i));
}
done.set(true);
gateway.stop();
}
@Test
@@ -256,6 +266,7 @@ public class TcpOutboundGatewayTests {
assertTrue(replies.remove("Reply" + i));
}
done.set(true);
gateway.stop();
}
@Test
@@ -390,6 +401,171 @@ public class TcpOutboundGatewayTests {
assertEquals(lastReceived.get().replace("Test", "Reply"), replies.get(0));
done.set(true);
assertEquals(0, TestUtils.getPropertyValue(gateway, "pendingReplies", Map.class).size());
gateway.stop();
}
@Test
public void testCachingFailover() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final CountDownLatch serverLatch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
while (!done.get()) {
Socket socket = server.accept();
while (!socket.isClosed()) {
try {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
String request = (String) ois.readObject();
logger.debug("Read " + request);
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
oos.writeObject("bar");
logger.debug("Replied to " + request);
serverLatch.countDown();
}
catch (IOException e) {
logger.debug("error on write " + e.getClass().getSimpleName());
socket.close();
}
}
}
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
// Failover
AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
TcpConnectionSupport mockConn1 = makeMockConnection();
when(factory1.getConnection()).thenReturn(mockConn1);
doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class));
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port);
factory2.setSerializer(new DefaultSerializer());
factory2.setDeserializer(new DefaultDeserializer());
factory2.setSoTimeout(10000);
factory2.setSingleUse(false);
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(factory1);
factories.add(factory2);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
// Cache
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2);
cachingFactory.start();
TcpOutboundGateway gateway = new TcpOutboundGateway();
gateway.setConnectionFactory(cachingFactory);
PollableChannel outputChannel = new QueueChannel();
gateway.setOutputChannel(outputChannel);
gateway.afterPropertiesSet();
gateway.start();
GenericMessage<String> message = new GenericMessage<String>("foo");
gateway.handleMessage(message);
Message<?> reply = outputChannel.receive(0);
assertNotNull(reply);
assertEquals("bar", reply.getPayload());
done.set(true);
gateway.stop();
verify(mockConn1).send(Mockito.any(Message.class));
}
@Test
public void testFailoverCached() throws Exception {
final int port = SocketUtils.findAvailableServerSocket();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicBoolean done = new AtomicBoolean();
final CountDownLatch serverLatch = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port);
latch.countDown();
while (!done.get()) {
Socket socket = server.accept();
while (!socket.isClosed()) {
try {
ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
String request = (String) ois.readObject();
logger.debug("Read " + request);
ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
oos.writeObject("bar");
logger.debug("Replied to " + request);
serverLatch.countDown();
}
catch (IOException e) {
logger.debug("error on write " + e.getClass().getSimpleName());
socket.close();
}
}
}
}
catch (Exception e) {
if (!done.get()) {
e.printStackTrace();
}
}
}
});
assertTrue(latch.await(10000, TimeUnit.MILLISECONDS));
// Cache
AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
TcpConnectionSupport mockConn1 = makeMockConnection();
when(factory1.getConnection()).thenReturn(mockConn1);
doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class));
CachingClientConnectionFactory cachingFactory1 = new CachingClientConnectionFactory(factory1, 1);
AbstractClientConnectionFactory factory2 = new TcpNetClientConnectionFactory("localhost", port);
factory2.setSerializer(new DefaultSerializer());
factory2.setDeserializer(new DefaultDeserializer());
factory2.setSoTimeout(10000);
factory2.setSingleUse(false);
CachingClientConnectionFactory cachingFactory2 = new CachingClientConnectionFactory(factory2, 1);
// Failover
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(cachingFactory1);
factories.add(cachingFactory2);
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
TcpOutboundGateway gateway = new TcpOutboundGateway();
gateway.setConnectionFactory(failoverFactory);
PollableChannel outputChannel = new QueueChannel();
gateway.setOutputChannel(outputChannel);
gateway.afterPropertiesSet();
gateway.start();
GenericMessage<String> message = new GenericMessage<String>("foo");
gateway.handleMessage(message);
Message<?> reply = outputChannel.receive(0);
assertNotNull(reply);
assertEquals("bar", reply.getPayload());
done.set(true);
gateway.stop();
verify(mockConn1).send(Mockito.any(Message.class));
}
public TcpConnectionSupport makeMockConnection() {
TcpConnectionSupport connection = mock(TcpConnectionSupport.class);
when(connection.isOpen()).thenReturn(true);
return connection;
}
}

42
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java
View File

@@ -25,6 +25,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
@@ -33,6 +34,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -350,4 +352,44 @@ public class CachingClientConnectionFactoryTests {
okToRun.set(false);
}
@Test
public void testCachedFailover() throws Exception {
// Failover
AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
List<AbstractClientConnectionFactory> factories = new ArrayList<AbstractClientConnectionFactory>();
factories.add(factory1);
factories.add(factory2);
TcpConnectionSupport mockConn1 = makeMockConnection();
TcpConnectionSupport mockConn2 = makeMockConnection();
when(factory1.getConnection()).thenReturn(mockConn1);
when(factory2.getConnection()).thenReturn(mockConn2);
when(factory1.isActive()).thenReturn(true);
when(factory2.isActive()).thenReturn(true);
doThrow(new IOException("fail")).when(mockConn1).send(Mockito.any(Message.class));
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) throws Throwable {
return null;
}
}).when(mockConn2).send(Mockito.any(Message.class));
FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
failoverFactory.start();
// Cache
CachingClientConnectionFactory cachingFactory = new CachingClientConnectionFactory(failoverFactory, 2);
cachingFactory.start();
TcpConnection conn1 = cachingFactory.getConnection();
GenericMessage<String> message = new GenericMessage<String>("foo");
conn1 = cachingFactory.getConnection();
conn1.send(message);
Mockito.verify(mockConn2).send(message);
}
public TcpConnectionSupport makeMockConnection() {
TcpConnectionSupport connection = mock(TcpConnectionSupport.class);
when(connection.isOpen()).thenReturn(true);
return connection;
}
}