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

INT-3163 Force Connection Close on Send Exception

Browse Source 
Previously, when an exception occurred on a send, the
connection was not forcibly closed. When using a
CachingClientConnectionFactory, this prevented the connection
(albeit stale) from being returned to the cache.

Perform a forced (physical) close whenever a send fails.

Add test cases for both Net and NIO implementations, using
a CCCF, to verify the connection is returned to the pool so
the closed state can be detected on the next retrieval, causing
a refresh.

Also, change the synchronization in the NIO send to
synchronize on the socketChannel, not the mapper (which is
shared).

JIRA: https://jira.springsource.org/browse/INT-3163
This commit is contained in:
Gary Russell
2013-10-06 13:01:22 +03:00
committed by Artem Bilan
parent 1fa9b92c0b
commit 8dca61c62a
4 changed files with 108 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -100,6 +100,7 @@ public class TcpNetConnection extends TcpConnectionSupport {
}
catch (Exception e) {
this.publishConnectionExceptionEvent(e);
this.closeConnection(true);
throw e;
}
this.afterSend(message);

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -134,7 +134,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
@SuppressWarnings("unchecked")
public void send(Message<?> message) throws Exception {
synchronized(this.getMapper()) {
synchronized(this.socketChannel) {
Object object = this.getMapper().fromMessage(message);
this.lastSend = System.currentTimeMillis();
try {
@@ -142,6 +142,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
catch (Exception e) {
this.publishConnectionExceptionEvent(e);
this.closeConnection(true);
throw e;
}
this.afterSend(message);

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

@@ -20,6 +20,8 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -27,6 +29,10 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
@@ -39,12 +45,16 @@ import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.ip.util.TestingUtilities;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
@@ -286,6 +296,84 @@ public class CachingClientConnectionFactoryTests {
verify(mockConn2).close();
}
@Test
public void testExceptionOnSendNet() throws Exception {
TcpConnectionSupport conn1 = mockedTcpNetConnection();
TcpConnectionSupport conn2 = mockedTcpNetConnection();
CachingClientConnectionFactory cccf = createCCCFWith2Connections(conn1, conn2);
doTestCloseOnSendError(conn1, conn2, cccf);
}
@Test
public void testExceptionOnSendNio() throws Exception {
TcpConnectionSupport conn1 = mockedTcpNioConnection();
TcpConnectionSupport conn2 = mockedTcpNioConnection();
CachingClientConnectionFactory cccf = createCCCFWith2Connections(conn1, conn2);
doTestCloseOnSendError(conn1, conn2, cccf);
}
private void doTestCloseOnSendError(TcpConnection conn1, TcpConnection conn2,
CachingClientConnectionFactory cccf) throws Exception {
TcpConnection cached1 = cccf.getConnection();
try {
cached1.send(new GenericMessage<String>("foo"));
fail("Expected IOException");
}
catch (IOException e) {
assertEquals("Foo", e.getMessage());
}
// Before INT-3163 this failed with a timeout - connection not returned to pool after failure on send()
TcpConnection cached2 = cccf.getConnection();
assertTrue(cached1.getConnectionId().contains(conn1.getConnectionId()));
assertTrue(cached2.getConnectionId().contains(conn2.getConnectionId()));
}
private CachingClientConnectionFactory createCCCFWith2Connections(TcpConnectionSupport conn1, TcpConnectionSupport conn2)
throws Exception {
AbstractClientConnectionFactory factory = mock(AbstractClientConnectionFactory.class);
when(factory.isRunning()).thenReturn(true);
when(factory.getConnection()).thenReturn(conn1, conn2);
CachingClientConnectionFactory cccf = new CachingClientConnectionFactory(factory, 1);
cccf.setConnectionWaitTimeout(1);
cccf.start();
return cccf;
}
private TcpConnectionSupport mockedTcpNetConnection() throws IOException {
Socket socket = mock(Socket.class);
when(socket.isClosed()).thenReturn(true); // closed when next retrieved
OutputStream stream = mock(OutputStream.class);
doThrow(new IOException("Foo")).when(stream).write(Mockito.any(byte[].class));
when(socket.getOutputStream()).thenReturn(stream);
TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
}
}, "foo");
conn.setMapper(new TcpMessageMapper());
conn.setSerializer(new ByteArrayCrLfSerializer());
return conn;
}
private TcpConnectionSupport mockedTcpNioConnection() throws Exception {
SocketChannel socketChannel = mock(SocketChannel.class);
new DirectFieldAccessor(socketChannel).setPropertyValue("open", false);
doThrow(new IOException("Foo")).when(socketChannel).write(Mockito.any(ByteBuffer.class));
when(socketChannel.socket()).thenReturn(mock(Socket.class));
TcpNioConnection conn = new TcpNioConnection(socketChannel, false, false, new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
}
}, "foo");
conn.setMapper(new TcpMessageMapper());
conn.setSerializer(new ByteArrayCrLfSerializer());
return conn;
}
private TcpConnectionSupport makeMockConnection(String name) {
return makeMockConnection(name, false);
}

31
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/ConnectionEventTests.java
View File

@@ -24,7 +24,8 @@ import static org.mockito.Mockito.mock;
import java.io.OutputStream;
import java.net.Socket;
import java.util.concurrent.atomic.AtomicReference;
import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.mockito.Mockito;
@@ -44,16 +45,17 @@ public class ConnectionEventTests {
@Test
public void test() throws Exception {
Socket socket = mock(Socket.class);
final AtomicReference<TcpConnectionEvent> theEvent = new AtomicReference<TcpConnectionEvent>();
final List<TcpConnectionEvent> theEvent = new ArrayList<TcpConnectionEvent>();
TcpNetConnection conn = new TcpNetConnection(socket, false, false, new ApplicationEventPublisher() {
public void publishEvent(ApplicationEvent event) {
theEvent.set((TcpConnectionEvent) event);
theEvent.add((TcpConnectionEvent) event);
}
}, "foo");
assertNotNull(theEvent.get());
assertTrue(theEvent.get() instanceof TcpConnectionOpenEvent);
assertTrue(theEvent.get().toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **OPENED**"));
assertTrue(theEvent.size() > 0);
assertNotNull(theEvent.get(0));
assertTrue(theEvent.get(0) instanceof TcpConnectionOpenEvent);
assertTrue(theEvent.get(0).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **OPENED**"));
@SuppressWarnings("unchecked")
Serializer<Object> serializer = mock(Serializer.class);
RuntimeException toBeThrown = new RuntimeException("foo");
@@ -65,16 +67,17 @@ public class ConnectionEventTests {
fail("Expected exception");
}
catch (Exception e) {}
assertNotNull(theEvent.get());
assertTrue(theEvent.get() instanceof TcpConnectionExceptionEvent);
assertTrue(theEvent.get().toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]"));
assertTrue(theEvent.get().toString().contains("cause=java.lang.RuntimeException: foo]"));
TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get();
assertTrue(theEvent.size() > 1);
assertNotNull(theEvent.get(1));
assertTrue(theEvent.get(1) instanceof TcpConnectionExceptionEvent);
assertTrue(theEvent.get(1).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "]"));
assertTrue(theEvent.get(1).toString().contains("cause=java.lang.RuntimeException: foo]"));
TcpConnectionExceptionEvent event = (TcpConnectionExceptionEvent) theEvent.get(1);
assertNotNull(event.getCause());
assertSame(toBeThrown, event.getCause());
conn.close();
assertNotNull(theEvent.get());
assertTrue(theEvent.get().toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
assertTrue(theEvent.size() > 2);
assertNotNull(theEvent.get(2));
assertTrue(theEvent.get(2).toString().endsWith("[factory=foo, connectionId=" + conn.getConnectionId() + "] **CLOSED**"));
}
}