diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index a1ddec1e0d..794cb3ef5d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -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()) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index 8e7a582142..f723ab18ac 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -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. */ diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java index aef6a600a4..63c5973f72 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java @@ -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 results = new HashSet(); 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 results = new HashSet(); 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 { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index 7331b309a6..f7b9b61a77 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -213,28 +213,29 @@ public class TcpOutboundGatewayTests { })); } Set replies = new HashSet(); + 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)); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index d776b12876..8f820c8e6d 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -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 results = new HashSet(); + 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 results = new HashSet(); 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, diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 47bf3eb743..8ed8343420 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -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 results = new HashSet(); 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); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 88d71686ee..3e4b926211 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -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) 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) responses.get(0)).getPayload())); + } scf.close(); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java index 7cc0a6dbfb..92b5e9bb35 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketUtils.java @@ -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();