From 390cc84c5b41e4375bb89b6007231bc7727bd112 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 10 Aug 2010 20:47:18 +0000 Subject: [PATCH] INT-1279 TCP ib and ob Gateways Using Connection Factories - namespace and docs to follow --- .../integration/ip/tcp/TcpInboundGateway.java | 89 +++++++ .../ip/tcp/TcpOutboundGateway.java | 183 +++++++++++++ .../ip/tcp/TcpReceivingChannelAdapter.java | 3 +- .../connection/AbstractConnectionFactory.java | 5 +- .../AbstractTcpConnectionInterceptor.java | 4 +- .../ip/tcp/connection/TcpListener.java | 3 +- .../ip/tcp/connection/TcpNetConnection.java | 8 +- .../ip/tcp/connection/TcpNioConnection.java | 8 +- .../ip/tcp/TcpInboundGatewayTests.java | 168 ++++++++++++ .../ip/tcp/TcpOutboundGatewayTests.java | 240 ++++++++++++++++++ .../tcp/TcpReceivingChannelAdapterTests.java | 2 - .../tcp/connection/HelloWorldInterceptor.java | 8 +- .../integration/ip/udp/MultiClientTests.java | 6 +- 13 files changed, 707 insertions(+), 20 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java new file mode 100644 index 0000000000..9620c93074 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpInboundGateway.java @@ -0,0 +1,89 @@ +/* + * 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; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.springframework.integration.Message; +import org.springframework.integration.gateway.AbstractMessagingGateway; +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnection; +import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.ip.tcp.connection.TcpSender; + +/** + * Inbound Gateway using a server connection factory - threading is controlled by the + * factory. For java.net connections, each socket can process only one message at a time. + * For java.nio connections, messages may be multiplexed but the client will need to + * provide correlation logic. If the client is a {@link TcpOutboundGateway} multiplexing + * is not used, but multiple concurrent connections can be used if the connection factory uses + * single-use connections. For true asynchronous bi-directional communication, a pair of + * inbound / outbound channel adapters should be used. + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpInboundGateway extends AbstractMessagingGateway implements TcpListener, TcpSender { + + protected AbstractServerConnectionFactory connectionFactory; + + private Map connections = new ConcurrentHashMap(); + + public boolean onMessage(Message message) { + Message reply = this.sendAndReceiveMessage(message); + String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + TcpConnection connection = connections.get(connectionId); + if (connection == null) { + logger.error("Connection " + connectionId + " not found"); + return false; + } + try { + connection.send(reply); + } catch (Exception e) { + logger.error("Failed to send reply", e); + } + return false; + } + + /** + * @return true if the associated connection factory is listening. + */ + public boolean isListening() { + return connectionFactory.isListening(); + + } + + /** + * + * @param connectionFactory the Connection Factory + */ + public void setConnectionFactory(AbstractServerConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + connectionFactory.registerListener(this); + connectionFactory.registerSender(this); + } + + public void addNewConnection(TcpConnection connection) { + connections.put(connection.getConnectionId(), connection); + } + + public void removeDeadConnection(TcpConnection connection) { + connections.remove(connection.getConnectionId()); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java new file mode 100644 index 0000000000..045f1303eb --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java @@ -0,0 +1,183 @@ +/* + * 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; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +import org.springframework.integration.Message; +import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.MessagingException; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnection; +import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.util.Assert; + + +/** + * TCP outbound gateway that uses a client connection factory. If the factory is configured + * for single-use connections, each request is sent on a new connection; if the factory does not use + * single use connections, each request is blocked until the previous response is received + * (or times out). Asynchronous requests/responses over the same connection are not + * supported - use a pair of outbound/inbound adapters for that use case. + * + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler + implements TcpSender, TcpListener { + + protected AbstractConnectionFactory connectionFactory; + + private Map pendingReplies = new ConcurrentHashMap(); + + private Semaphore semaphore = new Semaphore(1, true); + + private long replyTimeout = 10000; + + private long requestTimeout = 10000; + + @Override + protected Object handleRequestMessage(Message requestMessage) { + Assert.notNull(connectionFactory, this.getClass().getName() + + " requires a client connection factory"); + boolean haveSemaphore = false; + try { + boolean singleUseConnection = this.connectionFactory.isSingleUse(); + if (!singleUseConnection) { + logger.debug("trying semaphore"); + if (!this.semaphore.tryAcquire(this.requestTimeout, TimeUnit.MILLISECONDS)) { + throw new MessageTimeoutException(requestMessage, "Timed out waiting for connection"); + } + haveSemaphore = true; + logger.debug("got semaphore"); + } + TcpConnection connection = this.connectionFactory.getConnection(); + AsyncReply reply = new AsyncReply(); + pendingReplies.put(connection.getConnectionId(), reply); + if (logger.isDebugEnabled()) { + logger.debug("Added " + connection.getConnectionId()); + } + connection.send(requestMessage); + Message replyMessage = reply.getReply(); + if (replyMessage == null) { + throw new MessageTimeoutException(requestMessage, "Timed out waiting for response"); + } + if (logger.isDebugEnabled()) { + logger.debug("Respose " + replyMessage); + } + return replyMessage; + } catch (Exception e) { + if (e instanceof MessagingException) { + throw (MessagingException) e; + } + throw new MessagingException("Failed to send or receive", e); + } finally { + if (haveSemaphore) { + this.semaphore.release(); + logger.debug("released semaphore"); + } + } + } + + public boolean onMessage(Message message) { + String connectionId = (String) message.getHeaders().get(IpHeaders.CONNECTION_ID); + if (connectionId == null) { + logger.error("Cannot correlate response - no connection id"); + return false; + } + AsyncReply reply = pendingReplies.get(connectionId); + if (reply == null) { + logger.error("Cannot correlate response - no pending reply"); + return false; + } + reply.setReply(message); + return false; + } + + public boolean isListening() { + return false; + } + + public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { + Assert.isTrue(connectionFactory instanceof AbstractClientConnectionFactory, + this.getClass().getName() + " requires a client connection factory"); + this.connectionFactory = connectionFactory; + connectionFactory.registerListener(this); + connectionFactory.registerSender(this); + } + + public void addNewConnection(TcpConnection connection) { + // do nothing - no asynchronous multiplexing supported + } + + public void removeDeadConnection(TcpConnection connection) { + // do nothing - no asynchronous multiplexing supported + } + + /** + * @param replyTimeout the replyTimeout to set + */ + public void setReplyTimeout(long timeout) { + this.replyTimeout = timeout; + } + + /** + * @param requestTimeout the requestTimeout to set + */ + public void setRequestTimeout(long requestTimeout) { + this.requestTimeout = requestTimeout; + } + + /** + * Class used to coordinate the asynchronous reply to its request. + * @author Gary Russell + * @since 2.0 + * + */ + class AsyncReply { + private CountDownLatch latch; + private Message reply; + public AsyncReply() { + this.latch = new CountDownLatch(1); + } + /** + * Sender blocks here until the reply is received, or we time out + * @return The return message or null if we time out + * @throws Exception + */ + public Message getReply() throws Exception { + if (!this.latch.await(replyTimeout, TimeUnit.MILLISECONDS)) { + return null; + } + return this.reply; + } + public void setReply(Message reply) { + this.reply = reply; + latch.countDown(); + } + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java index 23ec752cca..3bbf1b9f9c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -46,8 +46,9 @@ public class TcpReceivingChannelAdapter protected ConnectionFactory serverConnectionFactory; - public void onMessage(Message message) { + public boolean onMessage(Message message) { sendMessage(message); + return false; } @Override diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index 4d3ba93764..1229fcf538 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -329,10 +329,13 @@ public abstract class AbstractConnectionFactory this.interceptorFactoryChain.getInterceptorFactories()) { TcpConnectionInterceptor wrapper = factory.getInterceptor(); wrapper.setTheConnection(connection); - // if no ultimate listener, register each wrapper in turn + // if no ultimate listener or sender, register each wrapper in turn if (this.listener == null) { connection.registerListener(wrapper); } + if (this.sender == null) { + connection.registerSender(wrapper); + } connection = wrapper; } return connection; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java index 116e81af19..079ae4b111 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java @@ -110,11 +110,11 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI return this.theConnection.isServer(); } - public void onMessage(Message message) { + public boolean onMessage(Message message) { if (this.tcpListener == null) { throw new NoListenerException("No listener registered for message reception"); } - this.tcpListener.onMessage(message); + return this.tcpListener.onMessage(message); } public void send(Message message) throws Exception { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java index c27cfbf040..c93b493a6f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java @@ -31,7 +31,8 @@ public interface TcpListener { /** * Called by a TCPConnection when a new message arrives. * @param message The message. + * @return true if the message was intercepted */ - public abstract void onMessage(Message message); + public abstract boolean onMessage(Message message); } \ No newline at end of file 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 abbe8e7fe1..a1ddec1e0d 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 @@ -103,6 +103,7 @@ public class TcpNetConnection extends AbstractTcpConnection { Message message = null; boolean okToRun = true; logger.debug("Reading..."); + boolean intercepted = false; while (okToRun) { try { message = this.mapper.toMessage(this); @@ -127,7 +128,7 @@ public class TcpNetConnection extends AbstractTcpConnection { logger.warn("Unexpected message - no inbound adapter registered with connection " + message); continue; } - listener.onMessage(message); + intercepted = listener.onMessage(message); } catch (Exception e) { if (e instanceof NoListenerException) { if (this.singleUse) { @@ -143,9 +144,10 @@ public class TcpNetConnection extends AbstractTcpConnection { } /* * For single use sockets, we close after receipt if we are on the client - * side, or the server side has no outbound adapter registered + * side, and the data was not intercepted, + * or the server side has no outbound adapter registered */ - if (this.singleUse && this.server && this.sender == null) { + if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) { logger.debug("Closing single use socket after inbound message " + this.connectionId); this.close(); okToRun = false; 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 3e7a4e9fce..6b20070ce8 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 @@ -191,9 +191,10 @@ public class TcpNioConnection extends AbstractTcpConnection { return; } + boolean intercepted = false; try { if (message != null) { - listener.onMessage(message); + intercepted = listener.onMessage(message); } } catch (Exception e) { if (e instanceof NoListenerException) { @@ -209,9 +210,10 @@ public class TcpNioConnection extends AbstractTcpConnection { } /* * For single use sockets, we close after receipt if we are on the client - * side, or the server side has no outbound adapter registered + * side, and the data was not intercepted, + * or the server side has no outbound adapter registered */ - if (this.singleUse && this.server && this.sender == null) { + if (this.singleUse && ((!this.server && !intercepted) || (this.server && this.sender == null))) { logger.debug("Closing single use cbannel after inbound message " + this.connectionId); this.close(); } 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 new file mode 100644 index 0000000000..aef6a600a4 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpInboundGatewayTests.java @@ -0,0 +1,168 @@ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.io.InputStream; +import java.net.Socket; + +import javax.net.SocketFactory; + +import org.junit.Test; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.ChannelResolver; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.handler.ServiceActivatingHandler; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; +import org.springframework.integration.ip.util.SocketUtils; + + +public class TcpInboundGatewayTests { + + @Test + public void testNetSingle() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + scf.setSingleUse(true); + TcpInboundGateway gateway = new TcpInboundGateway(); + gateway.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 200) { + fail("Failed to listen"); + } + } + final QueueChannel channel = new QueueChannel(); + gateway.setRequestChannel(channel); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); + handler.setChannelResolver(new ChannelResolver() { + public MessageChannel resolveChannelName(String channelName) { + 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()); + byte[] bytes = new byte[12]; + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test1\r\n", new String(bytes)); + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test2\r\n", new String(bytes)); + } + + @Test + public void testNetNotSingle() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + scf.setSingleUse(false); + TcpInboundGateway gateway = new TcpInboundGateway(); + gateway.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 200) { + fail("Failed to listen"); + } + } + final QueueChannel channel = new QueueChannel(); + gateway.setRequestChannel(channel); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); + 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()); + byte[] bytes = new byte[12]; + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test1\r\n", new String(bytes)); + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test2\r\n", new String(bytes)); + } + + @Test + public void testNioSingle() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + scf.setSingleUse(true); + TcpInboundGateway gateway = new TcpInboundGateway(); + gateway.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 200) { + fail("Failed to listen"); + } + } + final QueueChannel channel = new QueueChannel(); + gateway.setRequestChannel(channel); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); + handler.setChannelResolver(new ChannelResolver() { + public MessageChannel resolveChannelName(String channelName) { + 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()); + byte[] bytes = new byte[12]; + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test1\r\n", new String(bytes)); + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test2\r\n", new String(bytes)); + } + + @Test + public void testNioNotSingle() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + scf.setSingleUse(false); + TcpInboundGateway gateway = new TcpInboundGateway(); + gateway.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 200) { + fail("Failed to listen"); + } + } + final QueueChannel channel = new QueueChannel(); + gateway.setRequestChannel(channel); + ServiceActivatingHandler handler = new ServiceActivatingHandler(new Service()); + 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()); + byte[] bytes = new byte[12]; + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test1\r\n", new String(bytes)); + readFully(socket.getInputStream(), bytes); + assertEquals("Echo:Test2\r\n", new String(bytes)); + } + + private class Service { + @SuppressWarnings("unused") + public String serviceMethod(byte[] bytes) { + return "Echo:" + new String(bytes); + } + } + + private void readFully(InputStream is, byte[] buff) throws IOException { + for (int i = 0; i < buff.length; i++) { + buff[i] = (byte) is.read(); + } + } + +} 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 new file mode 100644 index 0000000000..9477759054 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -0,0 +1,240 @@ +/* + * 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; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; + +import javax.net.ServerSocketFactory; + +import org.junit.Test; +import org.springframework.commons.serializer.java.JavaStreamingConverter; +import org.springframework.integration.Message; +import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessageBuilder; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; +import org.springframework.integration.ip.util.SocketUtils; + + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpOutboundGatewayTests { + + @Test + public void testGoodNetSingle() { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + while (true) { + Socket socket = server.accept(); + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + Object in = ois.readObject(); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + oos.writeObject("Reply" + (i++)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + JavaStreamingConverter converter = new JavaStreamingConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.setSingleUse(true); + ccf.setPoolSize(10); + ccf.start(); + TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(ccf); + QueueChannel replyChannel = new QueueChannel(); + gateway.setRequiresReply(true); + gateway.setOutputChannel(replyChannel); + gateway.setReplyTimeout(60000); + gateway.setRequestTimeout(60000); + for (int i = 100; i < 200; i++) { + gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build()); + } + Set replies = new HashSet(); + for (int i = 100; i < 200; i++) { + Message m = replyChannel.receive(10000); + assertNotNull(m); + replies.add((String) m.getPayload()); + } + for (int i = 0; i < 100; i++) { + assertTrue(replies.remove("Reply" + i)); + } + } + + @Test + public void testGoodNetMultiplex() { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + Socket socket = server.accept(); + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + ois.readObject(); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + oos.writeObject("Reply" + (i++)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + JavaStreamingConverter converter = new JavaStreamingConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + ccf.start(); + TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(ccf); + QueueChannel replyChannel = new QueueChannel(); + gateway.setRequiresReply(true); + gateway.setOutputChannel(replyChannel); + for (int i = 100; i < 110; i++) { + gateway.handleMessage(MessageBuilder.withPayload("Test" + i).build()); + } + Set replies = new HashSet(); + for (int i = 100; i < 110; i++) { + Message m = replyChannel.receive(10000); + assertNotNull(m); + replies.add((String) m.getPayload()); + } + for (int i = 0; i < 10; i++) { + assertTrue(replies.remove("Reply" + i)); + } + } + + @Test + public void testGoodNetTimeout() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + Socket socket = server.accept(); + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + ois.readObject(); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + Thread.sleep(1000); + oos.writeObject("Reply" + (i++)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + JavaStreamingConverter converter = new JavaStreamingConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.setSingleUse(false); + ccf.start(); + final TcpOutboundGateway gateway = new TcpOutboundGateway(); + gateway.setConnectionFactory(ccf); + gateway.setRequestTimeout(1); + QueueChannel replyChannel = new QueueChannel(); + gateway.setRequiresReply(true); + gateway.setOutputChannel(replyChannel); + List> results = new ArrayList>(); + for (int i = 0; i < 2; i++) { + final int j = i; + results.add(Executors.newSingleThreadExecutor().submit(new Callable(){ + public Integer call() throws Exception { + gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build()); + return 0; + } + })); + } + Set replies = new HashSet(); + for (int i = 0; i < 2; i++) { + try { + results.get(i).get(); + } catch (InterruptedException e) { + } catch (ExecutionException e) { + if (i == 0) { + fail("Unexpected " + e.getMessage()); + } else if (i == 1) { + assertNotNull(e.getCause()); + assertTrue(e.getCause() instanceof MessageTimeoutException); + } + + continue; + } + if (i == 1) { + fail("Expected ExecutionException"); + } + Message m = replyChannel.receive(10000); + assertNotNull(m); + replies.add((String) m.getPayload()); + } + 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 a2ae51bd89..4e48feb70d 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 @@ -25,8 +25,6 @@ import java.io.InputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.net.Socket; -import java.net.SocketException; -import java.net.UnknownHostException; import java.util.HashSet; import java.util.LinkedList; import java.util.List; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java index f40ec9a571..1dd69b5fd9 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/HelloWorldInterceptor.java @@ -58,7 +58,7 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { } @Override - public void onMessage(Message message) { + public boolean onMessage(Message message) { if (!this.negotiated) { Object payload = message.getPayload(); if (this.isServer()) { @@ -67,7 +67,7 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { logger.debug("sending " + this.world); super.send(MessageBuilder.withPayload(world).build()); this.negotiated = true; - return; + return true; } catch (Exception e) { throw new MessagingException("Negotiation error", e); } @@ -84,10 +84,10 @@ public class HelloWorldInterceptor extends AbstractTcpConnectionInterceptor { throw new MessagingException("Negotiation error - expected '" + world + "' received " + payload); } - return; + return true; } } - super.onMessage(message); + return super.onMessage(message); } @Override diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java index 9b3097dad4..aa7ca8c4a7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java @@ -65,7 +65,7 @@ public class MultiClientTests { UnicastSendingMessageHandler sender = new UnicastSendingMessageHandler( "localhost", adapter.getPort()); while (true) { - Message message = queueIn.receive(); + Message message = queueIn.receive(); sender.handleMessage(message); } }}); @@ -110,7 +110,7 @@ public class MultiClientTests { SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1000), 10000); while (true) { - Message message = queueIn.receive(); + Message message = queueIn.receive(); sender.handleMessage(message); } }}); @@ -155,7 +155,7 @@ public class MultiClientTests { SocketUtils.findAvailableUdpSocket(adapter.getPort() + j + 1100), 10000); while (true) { - Message message = queueIn.receive(); + Message message = queueIn.receive(); sender.handleMessage(message); } }});