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

INT-1369 Improve Asynchrony of NIO Connections

Browse Source
This commit is contained in:
Gary Russell
2010-08-23 23:05:19 +00:00
parent 1fa165de3a
commit 4090905770
8 changed files with 119 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -62,7 +62,7 @@ public class TcpNetConnection extends AbstractTcpConnection {
}
@SuppressWarnings("unchecked")
public void send(Message<?> message) throws Exception {
public synchronized void send(Message<?> message) throws Exception {
Object object = mapper.fromMessage(message);
this.outputConverter.convert(object, this.socket.getOutputStream());
if (logger.isDebugEnabled())

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

@@ -58,8 +58,6 @@ public class TcpNioConnection extends AbstractTcpConnection {
private int maxMessageSize = 60 * 1024;
private boolean active = true;
private long lastRead;
private AtomicInteger executionControl = new AtomicInteger();
@@ -89,7 +87,6 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
private void doClose() {
this.active = false;
if (pipedOutputStream != null) {
try {
pipedOutputStream.close();
@@ -107,8 +104,10 @@ public class TcpNioConnection extends AbstractTcpConnection {
@SuppressWarnings("unchecked")
public void send(Message<?> message) throws Exception {
Object object = mapper.fromMessage(message);
this.outputConverter.convert(object, this.channelOutputStream);
synchronized(mapper) {
Object object = mapper.fromMessage(message);
this.outputConverter.convert(object, this.channelOutputStream);
}
}
public String getHostAddress() {
@@ -150,28 +149,42 @@ public class TcpNioConnection extends AbstractTcpConnection {
* sockets.
*/
public void run() {
logger.debug("Nio message assembler running...");
logger.trace("Nio message assembler running...");
try {
if (this.listener == null && !this.singleUse) {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
while (active) {
try {
while (dataAvailable()) {
convertAndSend();
try {
if (dataAvailable()) {
Message<?> message = convert();
if (dataAvailable()) {
// there is more data in the pipe; run another assembler
// to assemble the next message, while we send ours
this.executionControl.incrementAndGet();
this.taskExecutor.execute(this);
}
if (message != null) {
sendToChannel(message);
}
} catch (IOException e) {
logger.error("Unexpected exception, exiting...", e);
return;
}
// currently no more work to do
if (this.executionControl.decrementAndGet() < 0) {
break;
}
} catch (IOException e) {
logger.error("Unexpected exception, exiting...", e);
return;
}
this.executionControl.decrementAndGet();
} finally {
logger.debug("Nio message assembler exiting...");
logger.trace("Nio message assembler exiting...");
// Final check in case new data came in and the
// timing was such that we were the last assembler and
// a new one wasn't run
try {
if (dataAvailable()) {
checkForAssembler();
}
} catch (IOException e) {
logger.error("Exception when checking for assembler", e);
}
}
}
@@ -180,9 +193,15 @@ public class TcpNioConnection extends AbstractTcpConnection {
(this.pipedInputStream.available() > 0 || writingToPipe);
}
private synchronized void convertAndSend() throws IOException {
/**
* Blocks until a complete message has been assembled.
* Synchronized to avoid concurrency.
* @return The Message or null if no data is available.
* @throws IOException
*/
private synchronized Message<?> convert() throws IOException {
if (!dataAvailable()) {
return;
return null;
}
Message<?> message = null;
try {
@@ -202,9 +221,12 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
}
return;
return null;
}
return message;
}
private void sendToChannel(Message<?> message) {
boolean intercepted = false;
try {
if (message != null) {
@@ -242,11 +264,8 @@ public class TcpNioConnection extends AbstractTcpConnection {
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newSingleThreadExecutor();
}
if (this.executionControl.incrementAndGet() <= 1) {
// only execute run() if we don't already have one running
this.executionControl.set(1);
this.taskExecutor.execute(this);
}
// If there is no assembler running, start one
checkForAssembler();
rawBuffer.clear();
int len = socketChannel.read(rawBuffer);
if (len < 0) {
@@ -263,6 +282,18 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
private void checkForAssembler() {
synchronized(this.executionControl) {
if (this.executionControl.incrementAndGet() <= 1) {
// only execute run() if we don't already have one running
this.executionControl.set(1);
this.taskExecutor.execute(this);
} else {
this.executionControl.decrementAndGet();
}
}
}
/**
* Invoked by the factory when there is data to be read.
*/

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

@@ -1,11 +1,14 @@
package org.springframework.integration.ip.tcp;
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.util.HashSet;
import java.util.Set;
import javax.net.SocketFactory;
@@ -115,11 +118,15 @@ public class TcpInboundGatewayTests {
socket.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);
assertEquals("Echo:Test1\r\n", new String(bytes));
results.add(new String(bytes));
readFully(socket.getInputStream(), bytes);
assertEquals("Echo:Test2\r\n", new String(bytes));
results.add(new String(bytes));
System.out.println(results);
assertTrue(results.remove("Echo:Test1\r\n"));
assertTrue(results.remove("Echo:Test2\r\n"));
}
@Test
@@ -145,11 +152,14 @@ public class TcpInboundGatewayTests {
socket.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);
assertEquals("Echo:Test1\r\n", new String(bytes));
results.add(new String(bytes));
readFully(socket.getInputStream(), bytes);
assertEquals("Echo:Test2\r\n", new String(bytes));
results.add(new String(bytes));
assertTrue(results.remove("Echo:Test1\r\n"));
assertTrue(results.remove("Echo:Test2\r\n"));
}
private class Service {

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

@@ -213,28 +213,29 @@ public class TcpOutboundGatewayTests {
}));
}
Set<String> replies = new HashSet<String>();
int timeouts = 0;
for (int i = 0; i < 2; i++) {
try {
results[i].get();
} catch (InterruptedException e) {
} catch (ExecutionException e) {
if (i == 0) {
if (timeouts > 0) {
fail("Unexpected " + e.getMessage());
} else if (i == 1) {
} else {
assertNotNull(e.getCause());
assertTrue(e.getCause() instanceof MessageTimeoutException);
}
timeouts++;
continue;
}
if (i == 1) {
fail("Expected ExecutionException");
}
Message<?> m = replyChannel.receive(10000);
assertNotNull(m);
replies.add((String) m.getPayload());
}
if (timeouts < 1) {
fail("Expected ExecutionException");
}
for (int i = 0; i < 1; i++) {
assertTrue(replies.remove("Reply" + i));
}

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

@@ -104,16 +104,17 @@ public class TcpReceivingChannelAdapterTests {
QueueChannel channel = new QueueChannel();
adapter.setOutputChannel(channel);
Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
for (int i = 0; i < 100; i++) {
for (int i = 0; i < 1000; i++) {
socket.getOutputStream().write(("Test" + i + "\r\n").getBytes());
// if (i % 10 == 0) {
// Thread.sleep(1000);
// }
}
for (int i = 0; i < 100; i++) {
}
Set<String> results = new HashSet<String>();
for (int i = 0; i < 1000; i++) {
Message<?> message = channel.receive(10000);
assertNotNull(message);
assertEquals("Test" + i, new String((byte[]) message.getPayload()));
results.add(new String((byte[]) message.getPayload()));
}
for (int i = 0; i < 1000; i++) {
assertTrue(results.remove("Test" + i));
}
}
@@ -370,7 +371,7 @@ public class TcpReceivingChannelAdapterTests {
handler.setConnectionFactory(scf);
TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
adapter.setConnectionFactory(scf);
Executor te = Executors.newFixedThreadPool(100);
Executor te = Executors.newFixedThreadPool(10);
scf.setTaskExecutor(te);
scf.start();
QueueChannel channel = new QueueChannel();
@@ -475,12 +476,15 @@ public class TcpReceivingChannelAdapterTests {
assertEquals("world!", new ObjectInputStream(socket.getInputStream()).readObject());
new ObjectOutputStream(socket.getOutputStream()).writeObject("Test1");
new ObjectOutputStream(socket.getOutputStream()).writeObject("Test2");
Set<String> results = new HashSet<String>();
Message<?> message = channel.receive(10000);
assertNotNull(message);
assertEquals("Test1", message.getPayload());
results.add((String) message.getPayload());
message = channel.receive(10000);
assertNotNull(message);
assertEquals("Test2", message.getPayload());
results.add((String) message.getPayload());
assertTrue(results.contains("Test1"));
assertTrue(results.contains("Test2"));
}
private void singleNoOutboundInterceptorsGuts(final int port,

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

@@ -837,12 +837,15 @@ public class TcpSendingMessageHandlerTests {
assertTrue(latch.await(10, TimeUnit.SECONDS));
handler.handleMessage(MessageBuilder.withPayload("Test").build());
handler.handleMessage(MessageBuilder.withPayload("Test").build());
Set<String> results = new HashSet<String>();
Message<?> mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply1", mOut.getPayload());
results.add((String) mOut.getPayload());
mOut = channel.receive(10000);
assertNotNull(mOut);
assertEquals("Reply2", mOut.getPayload());
results.add((String) mOut.getPayload());
assertTrue(results.remove("Reply1"));
assertTrue(results.remove("Reply2"));
done.set(true);
}

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

@@ -115,17 +115,24 @@ public class TcpNioConnectionReadTests {
AbstractServerConnectionFactory scf = getConnectionFactory(port, converter,new TcpListener() {
public boolean onMessage(Message<?> message) {
responses.add(message);
try {
Thread.sleep(1000);
} catch (InterruptedException e) { }
semaphore.release();
return false;
}
});
int howMany = 2;
scf.setPoolSize(howMany + 5);
// Fire up the sender.
SocketUtils.testSendFragmented(port, false);
assertTrue(semaphore.tryAcquire(1, 10000, TimeUnit.MILLISECONDS));
assertEquals("Data", "xx",
new String(((Message<byte[]>) responses.get(0)).getPayload()));
assertEquals("Expected", 1, responses.size());
SocketUtils.testSendFragmented(port, howMany, false);
assertTrue(semaphore.tryAcquire(howMany, 20000, TimeUnit.MILLISECONDS));
assertEquals("Expected", howMany, responses.size());
for (int i = 0; i < howMany; i++) {
assertEquals("Data", "xx",
new String(((Message<byte[]>) responses.get(0)).getPayload()));
}
scf.close();
}

16
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java
View File

@@ -102,19 +102,21 @@ public class SocketUtils {
* Test for reassembly of completely fragmented message; sends
* 6 bytes 500ms apart.
*/
public static void testSendFragmented(final int port, final boolean noDelay) {
public static void testSendFragmented(final int port, final int howMany, final boolean noDelay) {
Thread thread = new Thread(new Runnable() {
public void run() {
try {
logger.debug("Connecting to " + port);
Socket socket = new Socket(InetAddress.getByName("localhost"), port);
OutputStream os = socket.getOutputStream();
writeByte(os, 0, noDelay);
writeByte(os, 0, noDelay);
writeByte(os, 0, noDelay);
writeByte(os, 2, noDelay);
writeByte(os, 'x', noDelay);
writeByte(os, 'x', noDelay);
for (int i = 0; i < howMany; i++) {
writeByte(os, 0, noDelay);
writeByte(os, 0, noDelay);
writeByte(os, 0, noDelay);
writeByte(os, 2, noDelay);
writeByte(os, 'x', noDelay);
writeByte(os, 'x', noDelay);
}
Thread.sleep(1000000000L); // wait forever, but we're a daemon
} catch (Exception e) {
e.printStackTrace();