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

INT-1707 Don't set SO_LINGER unless specified; close socket on send when appropriate, unless SO_LINGER=0; ensure any self-called close() is properly routed through any interceptors; suppress error on read when socket closed normally

Browse Source
This commit is contained in:
Gary Russell
2010-12-22 17:21:40 -05:00
parent abec48dce3
commit cfd61c4218
15 changed files with 444 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java
View File

@@ -69,15 +69,16 @@ public class TcpInboundGatewayTests {
return channel;
}
});
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test1\r\n".getBytes());
socket.getOutputStream().write("Test2\r\n".getBytes());
handler.handleMessage(channel.receive());
handler.handleMessage(channel.receive());
Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port);
socket1.getOutputStream().write("Test1\r\n".getBytes());
Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port);
socket2.getOutputStream().write("Test2\r\n".getBytes());
handler.handleMessage(channel.receive(1000));
handler.handleMessage(channel.receive(1000));
byte[] bytes = new byte[12];
readFully(socket.getInputStream(), bytes);
readFully(socket1.getInputStream(), bytes);
assertEquals("Echo:Test1\r\n", new String(bytes));
readFully(socket.getInputStream(), bytes);
readFully(socket2.getInputStream(), bytes);
assertEquals("Echo:Test2\r\n", new String(bytes));
}
@@ -134,19 +135,17 @@ public class TcpInboundGatewayTests {
return channel;
}
});
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test1\r\n".getBytes());
socket.getOutputStream().write("Test2\r\n".getBytes());
Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port);
socket1.getOutputStream().write("Test1\r\n".getBytes());
Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port);
socket2.getOutputStream().write("Test2\r\n".getBytes());
handler.handleMessage(channel.receive());
handler.handleMessage(channel.receive());
Set<String> results = new HashSet<String>();
byte[] bytes = new byte[12];
readFully(socket.getInputStream(), bytes);
results.add(new String(bytes));
readFully(socket.getInputStream(), bytes);
results.add(new String(bytes));
assertTrue(results.remove("Echo:Test1\r\n"));
assertTrue(results.remove("Echo:Test2\r\n"));
readFully(socket1.getInputStream(), bytes);
assertEquals("Echo:Test1\r\n", new String(bytes));
readFully(socket2.getInputStream(), bytes);
assertEquals("Echo:Test2\r\n", new String(bytes));
}
@Test
@@ -210,13 +209,14 @@ public class TcpInboundGatewayTests {
gateway.setRequestChannel(channel);
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService());
channel.subscribe(handler);
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.getOutputStream().write("Test1\r\n".getBytes());
socket.getOutputStream().write("Test2\r\n".getBytes());
Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port);
socket1.getOutputStream().write("Test1\r\n".getBytes());
Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port);
socket2.getOutputStream().write("Test2\r\n".getBytes());
byte[] bytes = new byte[errorMessage.length() + 2];
readFully(socket.getInputStream(), bytes);
readFully(socket1.getInputStream(), bytes);
assertEquals(errorMessage + "\r\n", new String(bytes));
readFully(socket.getInputStream(), bytes);
readFully(socket2.getInputStream(), bytes);
assertEquals(errorMessage + "\r\n", new String(bytes));
}

1
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

@@ -518,6 +518,7 @@ public class TcpReceivingChannelAdapterTests {
new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello");
assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject());
new ObjectOutputStream(socket.getOutputStream()).writeObject("Test1");
socket = SocketFactory.getDefault().createSocket("localhost", port);
new ObjectOutputStream(socket.getOutputStream()).writeObject("Hello");
assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject());

56
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java
View File

@@ -42,6 +42,8 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor {
private String hello = "Hello";
private String world = "world!";
private boolean closeReceived;
public HelloWorldInterceptor() {
}
@@ -86,24 +88,58 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor {
return true;
}
}
return super.onMessage(message);
try {
return super.onMessage(message);
} finally {
// on the server side, we don't want to close if we are expecting a response
if (!(this.isServer() && this.hasRealSender())) {
this.checkDeferredClose();
}
}
}
@Override
public void send(Message<?> message) throws Exception {
if (!this.negotiated) {
if (!this.isServer()) {
logger.debug("Sending " + hello);
super.send(MessageBuilder.withPayload(hello).build());
this.negotiationSemaphore.tryAcquire(this.timeout, TimeUnit.MILLISECONDS);
if (!this.negotiated) {
throw new MessagingException("Negotiation error");
try {
if (!this.negotiated) {
if (!this.isServer()) {
logger.debug("Sending " + hello);
super.send(MessageBuilder.withPayload(hello).build());
this.negotiationSemaphore.tryAcquire(this.timeout, TimeUnit.MILLISECONDS);
if (!this.negotiated) {
throw new MessagingException("Negotiation error");
}
}
}
super.send(message);
} finally {
this.checkDeferredClose();
}
super.send(message);
}
/**
* Defer the close until we've actually sent the data after negotiation
*/
@Override
public void close() {
if (this.negotiated) {
super.close();
return;
}
closeReceived = true;
logger.debug("Deferring close");
}
/**
* Execute the close, if deferred
*/
private void checkDeferredClose() {
if (this.closeReceived) {
logger.debug("Executing deferred close");
this.close();
}
}
}

83
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests-context.xml Normal file
View File

@@ -0,0 +1,83 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip-2.0.xsd">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketTestUtils" />
<int-ip:tcp-connection-factory id="inCFNet"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(9000)}"
so-timeout="1000"
single-use="true"
/>
<int-ip:tcp-inbound-gateway request-channel="echo"
connection-factory="inCFNet" />
<int-ip:tcp-connection-factory id="inCFNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(9100)}"
so-timeout="1000"
single-use="true"
using-nio="true"
/>
<int-ip:tcp-inbound-gateway request-channel="echo"
connection-factory="inCFNio" />
<int-ip:tcp-connection-factory id="inCFNetRst"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(9200)}"
so-timeout="1000"
single-use="true"
so-linger="0"
/>
<int-ip:tcp-inbound-gateway request-channel="echo"
connection-factory="inCFNetRst" />
<int-ip:tcp-connection-factory id="inCFNioRst"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(9300)}"
so-timeout="1000"
single-use="true"
using-nio="true"
so-linger="0"
/>
<int-ip:tcp-inbound-gateway request-channel="echo"
connection-factory="inCFNioRst" />
<int-ip:tcp-connection-factory id="inCFNetLinger"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(9400)}"
so-timeout="1000"
single-use="true"
so-linger="1000"
/>
<int-ip:tcp-inbound-gateway request-channel="echo"
connection-factory="inCFNetLinger" />
<int-ip:tcp-connection-factory id="inCFNioLinger"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(9500)}"
so-timeout="1000"
single-use="true"
using-nio="true"
so-linger="1000"
/>
<int-ip:tcp-inbound-gateway request-channel="echo"
connection-factory="inCFNioLinger" />
<int:service-activator input-channel="echo" ref="testService"/>
<bean id="testService" class="org.springframework.integration.ip.tcp.TestService"/>
</beans>

166
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SOLingerTests.java Normal file
View File

@@ -0,0 +1,166 @@
/*
* Copyright 2002-2010 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.connection;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.net.SocketTimeoutException;
import javax.net.SocketFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 2.0.2
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class SOLingerTests {
@Autowired
private AbstractServerConnectionFactory inCFNet;
@Autowired
private AbstractServerConnectionFactory inCFNio;
@Autowired
private AbstractServerConnectionFactory inCFNetRst;
@Autowired
private AbstractServerConnectionFactory inCFNioRst;
@Autowired
private AbstractServerConnectionFactory inCFNetLinger;
@Autowired
private AbstractServerConnectionFactory inCFNioLinger;
@Test
public void configOk() {}
@Test
public void finReceivedNet() {
finReceived(inCFNet);
}
@Test
public void finReceivedNio() {
finReceived(inCFNio);
}
@Test
public void rstReceivedNet() {
rstReceived(inCFNetRst);
}
@Test
public void rstReceivedNio() {
rstReceived(inCFNioRst);
}
@Test
public void finReceivedNetLinger() {
finReceived(inCFNetLinger);
}
@Test
public void finReceivedNioLinger() {
finReceived(inCFNioLinger);
}
private void finReceived(AbstractServerConnectionFactory inCF) {
int port = inCF.getPort();
int n = 0;
while (!inCF.isListening()) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
fail("Interrupted");
}
if (n++ > 100) {
fail("Failed to start");
}
}
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
String test = "Test\r\n";
socket.getOutputStream().write(test.getBytes());
byte[] buff = new byte[test.length() + 5];
readFully(socket.getInputStream(), buff);
assertEquals("echo:" + test, new String(buff));
n = socket.getInputStream().read();
// we expect an orderly close
assertEquals(-1, n);
} catch (Exception e) {
e.printStackTrace();
fail("Unexpected Exception " + e.getMessage());
}
}
private void rstReceived(AbstractServerConnectionFactory inCF) {
int port = inCF.getPort();
int n = 0;
while (!inCF.isListening()) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
fail("Interrupted");
}
if (n++ > 100) {
fail("Failed to start");
}
}
try {
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
socket.setSoTimeout(200);
String test = "Test\r\n";
socket.getOutputStream().write(test.getBytes());
byte[] buff = new byte[test.length() + 5];
readFully(socket.getInputStream(), buff);
assertEquals("echo:" + test, new String(buff));
try {
n = socket.getInputStream().read();
fail("Expected IOException");
} catch (IOException ioe) {
assertTrue(ioe instanceof SocketTimeoutException);
}
} catch (Exception e) {
e.printStackTrace();
fail("Unexpected Exception " + e.getMessage());
}
}
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();
}
}
}

7
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java
View File

@@ -19,6 +19,10 @@ import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.net.Socket;
import javax.net.SocketFactory;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.ip.IpHeaders;
@@ -68,7 +72,8 @@ public class TcpMessageMapperTests {
public void testToMessageSequence() throws Exception {
TcpMessageMapper mapper = new TcpMessageMapper();
TcpConnection connection = new AbstractTcpConnection(false) {
Socket socket = SocketFactory.getDefault().createSocket();
TcpConnection connection = new AbstractTcpConnection(socket, false) {
public void run() {
}
public void send(Message<?> message) throws Exception {

5
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -235,6 +235,7 @@ public class TcpNioConnectionReadTests {
SocketTestUtils.testSendLengthOverflow(port);
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
}
@@ -273,6 +274,7 @@ public class TcpNioConnectionReadTests {
SocketTestUtils.testSendStxEtxOverflow(port);
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
}
@@ -311,6 +313,7 @@ public class TcpNioConnectionReadTests {
SocketTestUtils.testSendCrLfOverflow(port);
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
}
@@ -349,6 +352,7 @@ public class TcpNioConnectionReadTests {
socket.close();
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
}
@@ -416,6 +420,7 @@ public class TcpNioConnectionReadTests {
socket.close();
whileOpen(semaphore, added);
assertEquals(1, added.size());
assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS));
assertTrue(removed.size() > 0);
scf.close();
}