INT-1008 Simple TCP Inbound Gateway - Initial Commit

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java

@@ -29,16 +29,18 @@ public abstract class IpHeaders {
 
 	private static final String PREFIX = MessageHeaders.PREFIX;
 
-	private static final String IP = "ip_";
+	private static final String IP = PREFIX + "ip_";
 
-	private static final String TCP = "tcp_";
+	private static final String TCP = IP + "tcp_";
 
-	private static final String UDP = "udp_";
+	private static final String UDP = IP + "udp_";
 
-	public static final String HOSTNAME = PREFIX + IP + "hostname";
+	public static final String HOSTNAME = IP + "hostname";
 
-	public static final String IP_ADDRESS = PREFIX + IP + "address"; 
+	public static final String IP_ADDRESS = IP + "address"; 
 
-	public static final String ACK_ADDRESS = PREFIX + "ackTo";
+	public static final String ACK_ADDRESS = IP + "ackTo";
+
+	public static final String REMOTE_PORT = TCP + "remote_port";
 
 }

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java

@@ -174,20 +174,20 @@ public abstract class IpAdapterParserUtils {
 	 * @return The value of the attribute or false.
 	 */
 	static Integer getMessageFormat(Element element) {
-		String useNio = element.getAttribute(IpAdapterParserUtils.MESSAGE_FORMAT);
-		if (!StringUtils.hasText(useNio)) {
+		String messageFormat = element.getAttribute(IpAdapterParserUtils.MESSAGE_FORMAT);
+		if (!StringUtils.hasText(messageFormat)) {
 			return MessageFormats.FORMAT_LENGTH_HEADER;
 		}
-		if (useNio.equals("length-header")) {
+		if (messageFormat.equals("length-header")) {
 			return MessageFormats.FORMAT_LENGTH_HEADER;
 		}
-		if (useNio.equals("stx-etx")) {
+		if (messageFormat.equals("stx-etx")) {
 			return MessageFormats.FORMAT_STX_ETX;
 		}
-		if (useNio.equals("crlf")) {
+		if (messageFormat.equals("crlf")) {
 			return MessageFormats.FORMAT_CRLF;
 		}
-		if (useNio.equals("custom")) {
+		if (messageFormat.equals("custom")) {
 			return MessageFormats.FORMAT_CUSTOM;
 		}
 		return MessageFormats.FORMAT_LENGTH_HEADER;

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpInboundGatewayParser.java

@@ -0,0 +1,61 @@
+/*
+ * 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.config;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.core.Conventions;
+import org.springframework.integration.adapter.config.AbstractRemotingGatewayParser;
+import org.springframework.integration.ip.tcp.SimpleTcpNetInboundGateway;
+import org.w3c.dom.Element;
+
+import sun.print.IPPPrintService;
+
+/**
+ * @author Gary Russell
+ *
+ */
+public class IpInboundGatewayParser extends AbstractRemotingGatewayParser {
+
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser#getBeanClass(org.w3c.dom.Element)
+	 */
+	@Override
+	protected Class getBeanClass(Element element) {
+		return SimpleTcpNetInboundGateway.class;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.adapter.config.AbstractRemotingGatewayParser#isEligibleAttribute(java.lang.String)
+	 */
+	@Override
+	protected boolean isEligibleAttribute(String attributeName) {
+		return !attributeName.equals(IpAdapterParserUtils.MESSAGE_FORMAT)
+				&& super.isEligibleAttribute(attributeName);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.adapter.config.AbstractRemotingGatewayParser#doPostProcess(org.springframework.beans.factory.support.BeanDefinitionBuilder, org.w3c.dom.Element)
+	 */
+	@Override
+	protected void doPostProcess(BeanDefinitionBuilder builder, Element element) {
+		builder.addPropertyValue(
+				Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT), 
+				IpAdapterParserUtils.getMessageFormat(element));
+	}
+
+
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java

@@ -29,6 +29,7 @@ public class IpNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 	public void init() {
 		this.registerBeanDefinitionParser("inbound-channel-adapter", new IpInboundChannelAdapterParser());
 		this.registerBeanDefinitionParser("outbound-channel-adapter", new IpOutboundChannelAdapterParser());
+		this.registerBeanDefinitionParser("inbound-gateway", new IpInboundGatewayParser());
 	}
 
 }

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java

@@ -181,4 +181,11 @@ public class NetSocketReader extends AbstractSocketReader {
 		return this.socket.getInetAddress();
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#getSocket()
+	 */
+	public Socket getSocket() {
+		return socket;
+	}
+
 }

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java

@@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp;
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.nio.channels.SocketChannel;
 
@@ -331,4 +332,11 @@ public class NioSocketReader extends AbstractSocketReader {
 		this.usingDirectBuffers = usingDirectBuffers;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.ip.tcp.SocketReader#getSocket()
+	 */
+	public Socket getSocket() {
+		return channel.socket();
+	}
+
 }

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java

@@ -0,0 +1,244 @@
+/*
+ * 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.net.Socket;
+import java.net.SocketException;
+
+import org.springframework.integration.adapter.MessageMappingException;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.gateway.AbstractMessagingGateway;
+
+/**
+ * Simple implementation of a TCP/IP inbound gateway; uses {@link java.net.Socket}
+ * and socket reader thread hangs on receive for response; therefore no multiplexing
+ * of incoming messages is supported. Delegates most of its work to a private
+ * subclass of {@link TcpNetReceivingChannelAdapter}, overriding the 
+ * processMessage() method.
+ * 
+ * Consequently, the pool size needs to be large enough to support the maximum
+ * number of concurrent connections expected.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
+	
+	private SocketMessageMapper mapper = new SocketMessageMapper();
+
+	private WriteCapableTcpNetReceivingChannelAdapter delegate;
+
+	private int port;
+
+	private int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
+
+	private int poolSize = 2;
+
+	private int receiveBufferSize = 2048;
+
+	private boolean soKeepAlive;
+
+	private int soReceiveBufferSize = -1;
+
+	private int soSendBufferSize = -1;
+
+	private int soTimeout = 0;
+
+	private String customSocketReaderClassName;
+	
+	private String customSocketWriterClassName;
+	
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.gateway.AbstractMessagingGateway#doStart()
+	 */
+	@Override
+	protected void doStart() {
+		super.doStart();
+		delegate.start();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.gateway.AbstractMessagingGateway#doStop()
+	 */
+	@Override
+	protected void doStop() {
+		super.doStop();
+		delegate.stop();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.gateway.AbstractMessagingGateway#onInit()
+	 */
+	@Override
+	protected void onInit() throws Exception {
+		delegate = new WriteCapableTcpNetReceivingChannelAdapter(port);
+		delegate.setMessageFormat(messageFormat);
+		delegate.setPoolSize(poolSize);
+		delegate.setReceiveBufferSize(receiveBufferSize);
+		delegate.setSoKeepAlive(soKeepAlive);
+		delegate.setSoReceiveBufferSize(soReceiveBufferSize);
+		delegate.setSoSendBufferSize(soSendBufferSize);
+		delegate.setSoTimeout(soTimeout);
+		delegate.setTaskScheduler(getTaskScheduler());
+		delegate.setCustomSocketReaderClassName(customSocketReaderClassName);
+		super.onInit();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.gateway.AbstractMessagingGateway#fromMessage(org.springframework.integration.core.Message)
+	 */
+	@Override
+	protected Object fromMessage(Message<?> message) {
+		throw new MessageMappingException("Cannot map a message to an object in this gateway");
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.integration.gateway.AbstractMessagingGateway#toMessage(java.lang.Object)
+	 */
+	@Override
+	protected Message<?> toMessage(Object object) {
+		try {
+			return mapper.toMessage((SocketReader) object);
+		} catch (Exception e) {
+			throw new MessageMappingException("Failed to map message", e);
+		}
+	}
+	
+
+	/**
+	 * @param port the port to set
+	 */
+	public void setPort(int port) {
+		this.port = port;
+	}
+
+	/**
+	 * @param messageFormat the messageFormat to set
+	 */
+	public void setMessageFormat(int messageFormat) {
+		this.messageFormat = messageFormat;
+	}
+
+	/**
+	 * @param poolSize the poolSize to set
+	 */
+	public void setPoolSize(int poolSize) {
+		this.poolSize = poolSize;
+	}
+
+	/**
+	 * @param receiveBufferSize the receiveBufferSize to set
+	 */
+	public void setReceiveBufferSize(int receiveBufferSize) {
+		this.receiveBufferSize = receiveBufferSize;
+	}
+
+	/**
+	 * @param soKeepAlive the soKeepAlive to set
+	 */
+	public void setSoKeepAlive(boolean soKeepAlive) {
+		this.soKeepAlive = soKeepAlive;
+	}
+
+	/**
+	 * @return the port
+	 */
+	public int getPort() {
+		return port;
+	}
+
+	/**
+	 * @param soReceiveBufferSize the soReceiveBufferSize to set
+	 */
+	public void setSoReceiveBufferSize(int soReceiveBufferSize) {
+		this.soReceiveBufferSize = soReceiveBufferSize;
+	}
+
+	/**
+	 * @param soSendBufferSize the soSendBufferSize to set
+	 */
+	public void setSoSendBufferSize(int soSendBufferSize) {
+		this.soSendBufferSize = soSendBufferSize;
+	}
+
+	/**
+	 * @param soTimeout the soTimeout to set
+	 */
+	public void setSoTimeout(int soTimeout) {
+		this.soTimeout = soTimeout;
+	}
+
+	/**
+	 * @param customSocketReaderClassName the customSocketReaderClassName to set
+	 */
+	public void setCustomSocketReaderClassName(String customSocketReaderClassName) {
+		this.customSocketReaderClassName = customSocketReaderClassName;
+	}
+
+	/**
+	 * @param customSocketWriterClassName the customSocketWriterClassName to set
+	 */
+	public void setCustomSocketWriterClassName(String customSocketWriterClassName) {
+		this.customSocketWriterClassName = customSocketWriterClassName;
+	}
+
+	private class WriteCapableTcpNetReceivingChannelAdapter extends TcpNetReceivingChannelAdapter {
+
+		/**
+		 * @param port
+		 */
+		public WriteCapableTcpNetReceivingChannelAdapter(int port) {
+			super(port);
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter#processMessage(org.springframework.integration.core.Message)
+		 */
+		@Override
+		protected void processMessage(NetSocketReader reader) {
+			Socket socket = reader.getSocket();
+			NetSocketWriter writer = new NetSocketWriter(socket);
+			writer.setMessageFormat(messageFormat);
+			Message<?> message = sendAndReceiveMessage(reader);
+			try {
+				writer.write(mapper.fromMessage(message));
+			} catch (Exception e) {
+				throw new MessageMappingException("Failed to map and send response", e);
+			}
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter#doStart()
+		 */
+		@Override
+		protected void doStart() {
+;			super.doStart();
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.integration.ip.tcp.AbstractTcpReceivingChannelAdapter#setSocketOptions(java.net.Socket)
+		 */
+		@Override
+		protected void setSocketOptions(Socket socket) throws SocketException {
+			super.setSocketOptions(socket);
+			if (soSendBufferSize > 0) {
+				socket.setSendBufferSize(soSendBufferSize);
+			}
+		}
+		
+	}
+}

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java

@@ -60,6 +60,7 @@ public class SocketMessageMapper implements
 			message = MessageBuilder.withPayload(payload)
 					.setHeader(IpHeaders.HOSTNAME, socketReader.getAddress().getHostName())
 					.setHeader(IpHeaders.IP_ADDRESS, socketReader.getAddress().getHostAddress())
+					.setHeader(IpHeaders.REMOTE_PORT, socketReader.getSocket().getPort())
 					.build();
 		}
 		return message;

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java

@@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp;
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.net.Socket;
 
 /**
  * General interface for assembling message data from a TCP/IP Socket.
@@ -49,4 +50,9 @@ public interface SocketReader {
 	 * @return The InetAddress.
 	 */
 	public InetAddress getAddress();
+
+	/**
+	 * @return the Socket
+	 */
+	public Socket getSocket();
 }

org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java

@@ -111,16 +111,27 @@ public class TcpNetReceivingChannelAdapter extends
 		while (true) {
 			try {
 				if (reader.assembleData()) {
-					Message<byte[]> message = mapper.toMessage(reader);
-					if (message != null) {
-						sendMessage(message);
-					}
+					processMessage(reader);
 				}
 			} catch (Exception e) {
+				logger.error("processMessage failed", e);
 				return;
 			}
 		}
 	}
+
+	/**
+	 * @param reader
+	 * @return
+	 * @throws Exception
+	 */
+	protected void processMessage(NetSocketReader reader)
+			throws Exception {
+		Message<byte[]> message = mapper.toMessage(reader);
+		if (message != null) {
+			sendMessage(message);
+		}
+	}
 	
 	@Override
 	protected void doStop() {
@@ -140,8 +151,10 @@ public class TcpNetReceivingChannelAdapter extends
 	@SuppressWarnings("unchecked")
 	public void setCustomSocketReaderClassName(
 			String customSocketReaderClassName) throws ClassNotFoundException {
-		this.customSocketReader = (Class<NetSocketReader>) Class
+		if (customSocketReaderClassName != null) {
+			this.customSocketReader = (Class<NetSocketReader>) Class
 				.forName(customSocketReaderClassName);
+		}
 	}
 
 }

org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd

@@ -28,13 +28,6 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="ipAdapterType">
-						<xsd:annotation>
-							<xsd:appinfo>
-								<tool:annotation kind="ref">
-									<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
-								</tool:annotation>
-							</xsd:appinfo>
-						</xsd:annotation>
 					<xsd:attribute name="pool-size" type="xsd:string" />
 					<xsd:attribute name="receive-buffer-size" type="xsd:string" />
 					<xsd:attribute name="multicast-address" type="xsd:string" />
@@ -67,7 +60,14 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T
 					<xsd:attribute name="ack-timeout" type="xsd:string" />
 					<xsd:attribute name="min-acks-for-success" type="xsd:string" />
 					<xsd:attribute name="time-to-live" type="xsd:string" />
-					<xsd:attribute name="custom-socket-writer-class-name" type="xsd:string" />
+					<xsd:attribute name="custom-socket-writer-class-name" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+If message-format = 'custom' you must provide a sub class of the appropriate type to implement
+the custom message format. See java docs for TcpNetSendingChannelAdapter and TcpNioSendingChannelAdapter. 							
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="so-linger" type="xsd:string" />
 					<xsd:attribute name="so-tcp-no-delay" type="xsd:string" />
 					<xsd:attribute name="so-traffic-class" type="xsd:string" />
@@ -76,28 +76,109 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T
 		</xsd:complexType>
 	</xsd:element>
 
+	<xsd:element name="inbound-gateway">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines an inbound Gateway for receiving and replying to incoming IP packets.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="gatewayType">
+					<xsd:attribute name="pool-size" type="xsd:string" />
+					<xsd:attribute name="receive-buffer-size" type="xsd:string" />
+					<xsd:attribute name="custom-socket-reader-class-name" type="xsd:string" >
+						<xsd:annotation>
+							<xsd:documentation>
+If message-format = 'custom' you must provide a sub class of the appropriate type to implement
+the custom message format. See java docs for TcpNetReceivingChannelAdapter and TcpNioReceivingChannelAdapter. 							
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>	
+					<xsd:attribute name="custom-socket-writer-class-name" type="xsd:string">
+						<xsd:annotation>
+							<xsd:documentation>
+If message-format = 'custom' you must provide a sub class of the appropriate type to implement
+the custom message format. See java docs for TcpNetSendingChannelAdapter and TcpNioSendingChannelAdapter. 							
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
+	</xsd:element>
+
 	<xsd:complexType name="ipAdapterType">
 		<xsd:annotation>
 			<xsd:documentation>
 	Common configuration for IP-based adapters.
 			</xsd:documentation>
 		</xsd:annotation>
+		<xsd:complexContent>
+			<xsd:extension base="common-attributes">
+				<xsd:attribute name="channel" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
+							</tool:annotation>
+						</xsd:appinfo>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="protocol">
+					<xsd:simpleType>
+						<xsd:restriction base="xsd:NMTOKEN">
+							<xsd:enumeration value="tcp" />
+							<xsd:enumeration value="udp" />
+						</xsd:restriction>
+					</xsd:simpleType>
+				</xsd:attribute>
+				<xsd:attribute name="check-length" type="xsd:string" />
+				<xsd:attribute name="multicast" type="xsd:string" />
+			</xsd:extension>
+		</xsd:complexContent>
+	</xsd:complexType>
+
+	<xsd:complexType name="gatewayType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines common configuration for gateway adapters.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:complexContent>
+			<xsd:extension base="common-attributes">
+				<xsd:attribute name="name" type="xsd:string"/>
+				<xsd:attribute name="request-channel" type="xsd:string" use="required">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
+							</tool:annotation>
+						</xsd:appinfo>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="reply-channel" type="xsd:string">
+					<xsd:annotation>
+						<xsd:appinfo>
+							<tool:annotation kind="ref">
+								<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
+							</tool:annotation>
+						</xsd:appinfo>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="request-timeout" type="xsd:string"/>
+				<xsd:attribute name="reply-timeout" type="xsd:string"/>
+				<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
+			</xsd:extension>
+		</xsd:complexContent>
+	</xsd:complexType>
+
+	<xsd:complexType name="common-attributes">
 		<xsd:attribute name="id" type="xsd:ID"/>
-		<xsd:attribute name="channel" type="xsd:string" />
-		<xsd:attribute name="protocol">
-			<xsd:simpleType>
-				<xsd:restriction base="xsd:NMTOKEN">
-					<xsd:enumeration value="tcp" />
-					<xsd:enumeration value="udp" />
-				</xsd:restriction>
-			</xsd:simpleType>
-		</xsd:attribute>
 		<xsd:attribute name="port" type="xsd:string" />
 		<xsd:attribute name="so-receive-buffer-size" type="xsd:string" />
 		<xsd:attribute name="so-send-buffer-size" type="xsd:string" />
 		<xsd:attribute name="so-timeout" type="xsd:string" />
-		<xsd:attribute name="check-length" type="xsd:string" />
-		<xsd:attribute name="multicast" type="xsd:string" />
 		<xsd:attribute name="using-nio" type="xsd:string" />
 		<xsd:attribute name="message-format">
 			<xsd:simpleType>
@@ -111,7 +192,6 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T
 		</xsd:attribute>
 		<xsd:attribute name="using-direct-buffers" type="xsd:string" />
 		<xsd:attribute name="so-keep-alive" type="xsd:string" />
-		
 	</xsd:complexType>
 
 </xsd:schema>

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java

@@ -16,7 +16,6 @@
 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;

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests-context.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:ip="http://www.springframework.org/schema/integration/ip"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.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.xsd">
+
+	<beans:bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
+
+	<ip:inbound-gateway id="tcpGateway"
+		port="#{tcpIpUtils.findAvailableServerSocket(5200)}"
+		request-channel="toSA"
+		reply-channel="fromSA"
+		message-format="crlf" />
+
+	<channel id="toSA" />
+	
+	<channel id="fromSA">
+		<queue capacity="1"/>
+	</channel>
+	
+	<service-activator id="SA" 
+		input-channel="toSA"
+		output-channel="fromSA"
+		ref="service"
+		method="test"
+	/>
+	
+	<beans:bean id="service" class="org.springframework.integration.ip.tcp.TestService" />
+
+</beans:beans>

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGatewayTests.java

@@ -0,0 +1,58 @@
+/*
+ * 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.assertEquals;
+
+import java.net.Socket;
+
+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
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class SimpleTcpNetInboundGatewayTests {
+	
+	@Autowired
+	SimpleTcpNetInboundGateway gateway;
+	
+	@Test
+	public void test1() throws Exception {
+		Thread.sleep(2000);
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", gateway.getPort());
+		String greetings = "Hello World!";
+		socket.getOutputStream().write((greetings + "\r\n").getBytes());
+		StringBuilder sb = new StringBuilder();
+		int c;
+		while (true) {
+			c = socket.getInputStream().read();
+			sb.append((char) c);
+			if (c == '\n') {
+				break;
+			}
+		}
+		assertEquals("echo:" + greetings + "\r\n", sb.toString());
+	}
+}

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java

@@ -19,6 +19,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
 import java.net.InetAddress;
+import java.net.Socket;
 import java.net.UnknownHostException;
 
 import org.junit.Test;
@@ -50,6 +51,8 @@ public class SocketMessageMapperTests {
 				.getHeaders().get(IpHeaders.HOSTNAME));
 		assertEquals(InetAddress.getLocalHost().getHostAddress(), message
 				.getHeaders().get(IpHeaders.IP_ADDRESS));
+		assertEquals(0, message
+				.getHeaders().get(IpHeaders.REMOTE_PORT));
 	}
 
 	/**
@@ -92,6 +95,13 @@ public class SocketMessageMapperTests {
 			return false;
 		}
 
+		/* (non-Javadoc)
+		 * @see org.springframework.integration.ip.tcp.SocketReader#getSocket()
+		 */
+		public Socket getSocket() {
+			return new Socket();
+		}
+
 		
 	}
 	

org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TestService.java

@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * Simple echo service.
+ * 
+ * @author Gary Russell
+ *
+ */
+public class TestService {
+
+	public String test(byte[] bytes) {
+		return "echo:" + new String(bytes);
+	}
+}