diff --git a/org.springframework.integration.ip/.project b/org.springframework.integration.ip/.project index ed78e2585a..3b05a20b74 100644 --- a/org.springframework.integration.ip/.project +++ b/org.springframework.integration.ip/.project @@ -15,8 +15,14 @@ + + org.springframework.ide.eclipse.core.springbuilder + + + + org.springframework.ide.eclipse.core.springnature org.maven.ide.eclipse.maven2Nature org.eclipse.jdt.core.javanature diff --git a/org.springframework.integration.ip/.settings/org.springframework.ide.eclipse.beans.core.prefs b/org.springframework.integration.ip/.settings/org.springframework.ide.eclipse.beans.core.prefs new file mode 100644 index 0000000000..d6d6b0ae2c --- /dev/null +++ b/org.springframework.integration.ip/.settings/org.springframework.ide.eclipse.beans.core.prefs @@ -0,0 +1,4 @@ +#Fri Apr 30 21:58:37 MDT 2010 +eclipse.preferences.version=1 +org.springframework.ide.eclipse.beans.core.ignoreMissingNamespaceHandler=false +org.springframework.ide.eclipse.beans.core.loadNamespaceHandlerFromClasspath=true diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index c821439a0f..c93e799071 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -20,6 +20,7 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.core.Conventions; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.ip.tcp.MessageFormats; import org.springframework.util.StringUtils; @@ -105,6 +106,23 @@ public abstract class IpAdapterParserUtils { } } + /** + * @param element + * @param builder + * @param parserContext + */ + public static void addHostAndPortToConstructor(Element element, + BeanDefinitionBuilder builder, ParserContext parserContext) { + String host = element.getAttribute(IpAdapterParserUtils.HOST); + if (!StringUtils.hasText(host)) { + parserContext.getReaderContext().error(IpAdapterParserUtils.HOST + + " is required for IP outbound channel adapters", element); + } + builder.addConstructorArgValue(host); + String port = IpAdapterParserUtils.getPort(element, parserContext); + builder.addConstructorArgValue(port); + } + /** * Asserts that a protocol attribute (udp or tcp) is supplied, * @param element @@ -193,6 +211,29 @@ public abstract class IpAdapterParserUtils { return MessageFormats.FORMAT_LENGTH_HEADER; } + /** + * @param element + * @param builder + */ + public static void addOutboundTcpAttributes(Element element, + BeanDefinitionBuilder builder) { + builder.addPropertyValue( + Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT), + IpAdapterParserUtils.getMessageFormat(element)); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.CUSTOM_SOCKET_WRITER_CLASS_NAME); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.USING_DIRECT_BUFFERS); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_KEEP_ALIVE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_LINGER); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_TCP_NODELAY); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_TRAFFIC_CLASS); + } + /** * Sets the common port attributes on the bean being built (timeout, receive buffer size, * send buffer size). diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java index f8173f546c..00f5dba003 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java @@ -30,6 +30,7 @@ public class IpNamespaceHandler extends AbstractIntegrationNamespaceHandler { this.registerBeanDefinitionParser("inbound-channel-adapter", new IpInboundChannelAdapterParser()); this.registerBeanDefinitionParser("outbound-channel-adapter", new IpOutboundChannelAdapterParser()); this.registerBeanDefinitionParser("inbound-gateway", new IpInboundGatewayParser()); + this.registerBeanDefinitionParser("outbound-gateway", new IpOutboundGatewayParser()); } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java index 8204b7c5b5..933184dc40 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java @@ -19,7 +19,6 @@ package org.springframework.integration.ip.config; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.core.Conventions; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler; @@ -48,23 +47,6 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt return builder.getBeanDefinition(); } - /** - * @param element - * @param builder - * @param parserContext - */ - private void addHostAndPortToConstructor(Element element, - BeanDefinitionBuilder builder, ParserContext parserContext) { - String host = element.getAttribute(IpAdapterParserUtils.HOST); - if (!StringUtils.hasText(host)) { - parserContext.getReaderContext().error(IpAdapterParserUtils.HOST - + " is required for IP outbound channel adapters", element); - } - builder.addConstructorArgValue(host); - String port = IpAdapterParserUtils.getPort(element, parserContext); - builder.addConstructorArgValue(port); - } - /** * @param element * @param parserContext @@ -87,7 +69,7 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt builder = BeanDefinitionBuilder .genericBeanDefinition(UnicastSendingMessageHandler.class); } - addHostAndPortToConstructor(element, builder, parserContext); + IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext); IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CHECK_LENGTH, true); IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, @@ -135,22 +117,8 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt builder = BeanDefinitionBuilder .genericBeanDefinition(TcpNioSendingMessageHandler.class); } - addHostAndPortToConstructor(element, builder, parserContext); - builder.addPropertyValue( - Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT), - IpAdapterParserUtils.getMessageFormat(element)); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IpAdapterParserUtils.CUSTOM_SOCKET_WRITER_CLASS_NAME); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IpAdapterParserUtils.USING_DIRECT_BUFFERS); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IpAdapterParserUtils.SO_KEEP_ALIVE); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IpAdapterParserUtils.SO_LINGER); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IpAdapterParserUtils.SO_TCP_NODELAY); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, - IpAdapterParserUtils.SO_TRAFFIC_CLASS); + IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext); + IpAdapterParserUtils.addOutboundTcpAttributes(element, builder); return builder; } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java new file mode 100644 index 0000000000..02d85ad648 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java @@ -0,0 +1,52 @@ +/* + * 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.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractConsumerEndpointParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; + +/** + * Parser for the <outbound-gateway> element of the integration 'jms' namespace. + * + * @author Gary Russell + */ +public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser { + + @Override + protected String getInputChannelAttributeName() { + return "request-channel"; + } + + @Override + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway"); + IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext); + IpAdapterParserUtils.addCommonSocketOptions(builder, element); + IpAdapterParserUtils.addOutboundTcpAttributes(element, builder); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); + return builder; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java index c6a311c929..532b3e0d1d 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java @@ -111,7 +111,7 @@ public abstract class AbstractTcpSendingMessageHandler extends } writer.write(bytes); } catch (Exception e) { - writer = null; + this.writer = null; if (e instanceof MessageMappingException) { throw (MessageMappingException) e; } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java index b4a7536cdd..a109abc1e9 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java @@ -59,9 +59,9 @@ public class NetSocketReader extends AbstractSocketReader { if (logger.isDebugEnabled()) { logger.debug("Message length is " + messageLength); } - if (messageLength > maxMessageSize) { + if (messageLength > this.maxMessageSize) { throw new IOException("Message length " + messageLength + - " exceeds max message length: " + maxMessageSize); + " exceeds max message length: " + this.maxMessageSize); } byte[] messagePart = new byte[messageLength]; read(messagePart); @@ -77,7 +77,7 @@ public class NetSocketReader extends AbstractSocketReader { InputStream inputStream = socket.getInputStream(); if (inputStream.read() != STX) throw new MessageMappingException("Expected STX to begin message"); - byte[] buffer = new byte[maxMessageSize]; + byte[] buffer = new byte[this.maxMessageSize]; int n = 0; int bite; while ((bite = inputStream.read()) != ETX) { @@ -86,9 +86,9 @@ public class NetSocketReader extends AbstractSocketReader { throw new IOException("Socket Closed"); } buffer[n++] = (byte) bite; - if (n >= maxMessageSize) { + if (n >= this.maxMessageSize) { throw new IOException("ETX not found before max message length: " - + maxMessageSize); + + this.maxMessageSize); } } assembledData = new byte[n]; @@ -102,7 +102,7 @@ public class NetSocketReader extends AbstractSocketReader { @Override protected boolean assembleDataCrLfFormat() throws IOException { InputStream inputStream = socket.getInputStream(); - byte[] buffer = new byte[maxMessageSize]; + byte[] buffer = new byte[this.maxMessageSize]; int n = 0; int bite; while (true) { @@ -114,9 +114,9 @@ public class NetSocketReader extends AbstractSocketReader { if (n > 0 && bite == '\n' && buffer[n-1] == '\r') break; buffer[n++] = (byte) bite; - if (n >= maxMessageSize) { + if (n >= this.maxMessageSize) { throw new IOException("CRLF not found before max message length: " - + maxMessageSize); + + this.maxMessageSize); } }; assembledData = new byte[n-1]; diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java index b23ae0777c..ffef2171a7 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java @@ -15,13 +15,12 @@ */ package org.springframework.integration.ip.tcp; -import java.lang.reflect.Constructor; import java.net.Socket; import java.net.SocketException; -import org.springframework.beans.BeanUtils; import org.springframework.integration.core.Message; import org.springframework.integration.gateway.AbstractMessagingGateway; +import org.springframework.integration.ip.util.SocketIoUtils; import org.springframework.integration.message.MessageMappingException; /** @@ -39,88 +38,71 @@ import org.springframework.integration.message.MessageMappingException; */ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { - private SocketMessageMapper mapper = new SocketMessageMapper(); + protected SocketMessageMapper mapper = new SocketMessageMapper(); - private WriteCapableTcpNetReceivingChannelAdapter delegate; + protected WriteCapableTcpNetReceivingChannelAdapter delegate; - private int port; + protected int port; - private int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER; + protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER; - private int poolSize = 2; + protected int poolSize = 2; - private int receiveBufferSize = 2048; + protected int receiveBufferSize = 2048; - private boolean soKeepAlive; + protected boolean soKeepAlive; - private int soReceiveBufferSize = -1; + protected int soReceiveBufferSize = -1; - private int soSendBufferSize = -1; + protected int soSendBufferSize = -1; - private int soTimeout = 0; + protected int soTimeout = 0; - private String customSocketReaderClassName; + protected String customSocketReaderClassName; - private Class customSocketWriter; + protected Class customSocketWriterClass; - - /* (non-Javadoc) - * @see org.springframework.integration.gateway.AbstractMessagingGateway#doStart() - */ @Override protected void doStart() { super.doStart(); - delegate.start(); + this.delegate.start(); } - /* (non-Javadoc) - * @see org.springframework.integration.gateway.AbstractMessagingGateway#doStop() - */ @Override protected void doStop() { super.doStop(); - delegate.stop(); + this.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); + this.delegate = new WriteCapableTcpNetReceivingChannelAdapter(port); + this.delegate.setMessageFormat(messageFormat); + this.delegate.setPoolSize(poolSize); + this.delegate.setReceiveBufferSize(receiveBufferSize); + this.delegate.setSoKeepAlive(soKeepAlive); + this.delegate.setSoReceiveBufferSize(soReceiveBufferSize); + this.delegate.setSoSendBufferSize(soSendBufferSize); + this.delegate.setSoTimeout(soTimeout); + this.delegate.setTaskScheduler(getTaskScheduler()); + this.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); + return this.mapper.toMessage((SocketReader) object); } catch (Exception e) { throw new MessageMappingException("Failed to map message", e); } } - /** * @param port the port to set */ @@ -160,7 +142,7 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { * @return the port */ public int getPort() { - return port; + return this.port; } /** @@ -192,16 +174,16 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { } /** - * @param customSocketWriter the customSocketWriter to set + * @param customSocketWriterClassName the customSocketWriterClassName to set * @throws ClassNotFoundException */ @SuppressWarnings("unchecked") public void setCustomSocketWriterClassName( String customSocketWriterClassName) throws ClassNotFoundException { if (customSocketWriterClassName != null) { - this.customSocketWriter = (Class) Class + this.customSocketWriterClass = (Class) Class .forName(customSocketWriterClassName); - if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriter))) { + if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) { throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter"); } } @@ -216,43 +198,24 @@ public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway { 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 = null; - try { - if (messageFormat == MessageFormats.FORMAT_CUSTOM){ - Constructor ctor = customSocketWriter.getConstructor(Socket.class); - writer = BeanUtils.instantiateClass(ctor, socket); - } else { - writer = new NetSocketWriter(socket); - } - } catch (Exception e) { - throw new MessageMappingException("Error creating SocketWriter", e); - } - writer.setMessageFormat(messageFormat); Message message = sendAndReceiveMessage(reader); + NetSocketWriter writer = SocketIoUtils.createNetWriter(this.messageFormat, + customSocketWriterClass, socket); try { - writer.write(mapper.fromMessage(message)); + writer.write(this.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); diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java new file mode 100644 index 0000000000..2363618b4a --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java @@ -0,0 +1,210 @@ +/* + * 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 org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessagingException; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.ip.util.SocketIoUtils; + +/** + * Simple TCP outbound gateway; delegates write to a {@link TcpNetSendingMessageHandler} + * then blocks on read of same socket. Uses {@link java.net.Socket} and the client + * thread is dedicated to a request/response pair. No multiplexing of requests + * over the outbound socket are supported. This class is thread safe in that + * if multiple clients attempt to send a message, they will be blocked until + * any existing request/response is processed. + * + * @author Gary Russell + * @since 2.0 + */ +public class SimpleTcpNetOutboundGateway extends + AbstractReplyProducingMessageHandler { + + + protected TcpNetSendingMessageHandler handler; + + protected Class customSocketReaderClass; + + protected int messageFormat; + + protected int maxMessageSize = 2048; + + protected int soReceiveBufferSize = -1; + + protected NetSocketReader reader; + + /** + * Constructs a SimpleTcpNetOutboundGateway that sends data to the + * specified host and port, and waits for a response. + * + * @param host The host. + * @param port The port. + */ + public SimpleTcpNetOutboundGateway(String host, int port) { + handler = new TcpNetSendingMessageHandler(host, port); + } + + /** + * Synchronized to prevent multiplexing requests over the same socket. + */ + @Override + protected synchronized Object handleRequestMessage(Message requestMessage) { + this.handler.handleMessage(requestMessage); + if (this.reader == null) { + Socket socket = this.handler.getSocket(); + this.reader = SocketIoUtils.createNetReader(this.messageFormat, + this.customSocketReaderClass, socket, this.maxMessageSize, + this.soReceiveBufferSize); + } + try { + this.reader.assembleData(); // Net... always returns true + byte[] bytes = this.reader.getAssembledData(); + return bytes; + } catch (Exception e) { + this.reader = null; + throw new MessagingException(requestMessage, e); + } + } + + /** + * @param obj + * @return + * @see java.lang.Object#equals(java.lang.Object) + */ + public boolean equals(Object obj) { + return handler.equals(obj); + } + + /** + * @return + * @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#getPort() + */ + public int getPort() { + return handler.getPort(); + } + + /** + * @return + * @see java.lang.Object#hashCode() + */ + public int hashCode() { + return handler.hashCode(); + } + + /** + * @param customSocketWriterClassName + * @throws ClassNotFoundException + * @see org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler#setCustomSocketWriterClassName(java.lang.String) + */ + public void setCustomSocketWriterClassName( + String customSocketWriterClassName) throws ClassNotFoundException { + handler.setCustomSocketWriterClassName(customSocketWriterClassName); + } + + /** + * @param messageFormat + * @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setMessageFormat(int) + */ + public void setMessageFormat(int messageFormat) { + handler.setMessageFormat(messageFormat); + this.messageFormat = messageFormat; + } + + /** + * @param soKeepAlive + * @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoKeepAlive(boolean) + */ + public void setSoKeepAlive(boolean soKeepAlive) { + handler.setSoKeepAlive(soKeepAlive); + } + + /** + * @param soLinger + * @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoLinger(int) + */ + public void setSoLinger(int soLinger) { + handler.setSoLinger(soLinger); + } + + /** + * @param size + * @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#setSoReceiveBufferSize(int) + */ + public void setSoReceiveBufferSize(int size) { + this.soReceiveBufferSize = size; + } + + /** + * @param size + * @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#setSoSendBufferSize(int) + */ + public void setSoSendBufferSize(int size) { + handler.setSoSendBufferSize(size); + } + + /** + * @param soTcpNoDelay + * @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoTcpNoDelay(boolean) + */ + public void setSoTcpNoDelay(boolean soTcpNoDelay) { + handler.setSoTcpNoDelay(soTcpNoDelay); + } + + /** + * @param timeout + * @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#setSoTimeout(int) + */ + public void setSoTimeout(int timeout) { + handler.setSoTimeout(timeout); + } + + /** + * @param soTrafficClass + * @see org.springframework.integration.ip.tcp.AbstractTcpSendingMessageHandler#setSoTrafficClass(int) + */ + public void setSoTrafficClass(int soTrafficClass) { + handler.setSoTrafficClass(soTrafficClass); + } + + /** + * @param customSocketReaderClass the customSocketReader to set + * @throws ClassNotFoundException + */ + @SuppressWarnings("unchecked") + public void setCustomSocketReaderClassName( + String customSocketReaderClassName) throws ClassNotFoundException { + if (customSocketReaderClassName != null) { + this.customSocketReaderClass = (Class) Class + .forName(customSocketReaderClassName); + if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) { + throw new IllegalArgumentException("Custom socket reader must be of type NetSocketReader"); + } + } + } + + /** + * Specify the Spring Integration reply channel. If this property is not + * set the gateway will check for a 'replyChannel' header on the request. + */ + public void setReplyChannel(MessageChannel replyChannel) { + this.setOutputChannel(replyChannel); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java index 2ae0003909..cada12e902 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java @@ -35,4 +35,9 @@ public interface SocketWriter { */ void write(byte[] bytes) throws IOException; + /** + * @param messageFormat the messageFormat to set + */ + public void setMessageFormat(int messageFormat); + } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java index d66dc76241..5a1b100b3b 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java @@ -16,15 +16,13 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; -import java.lang.reflect.Constructor; import java.net.ServerSocket; import java.net.Socket; import javax.net.ServerSocketFactory; -import org.springframework.beans.BeanUtils; import org.springframework.integration.core.Message; -import org.springframework.integration.message.MessageMappingException; +import org.springframework.integration.ip.util.SocketIoUtils; /** * Tcp Receiving Channel adapter that uses a {@link java.net.Socket}. Each @@ -38,7 +36,7 @@ public class TcpNetReceivingChannelAdapter extends AbstractTcpReceivingChannelAdapter { protected ServerSocket serverSocket; - protected Class customSocketReader; + protected Class customSocketReaderClass; /** * Constructs a TcpNetReceivingChannelAdapter that listens on the port. * @param port The port. @@ -93,21 +91,9 @@ public class TcpNetReceivingChannelAdapter extends * @param socket */ protected void handleSocket(Socket socket) { - NetSocketReader reader = null; - if (messageFormat == MessageFormats.FORMAT_CUSTOM) { - try { - Constructor ctor = - customSocketReader.getConstructor(Socket.class); - reader = BeanUtils.instantiateClass(ctor, socket); - } catch (Exception e) { - throw new MessageMappingException("Failed to instantiate custom reader", e); - } - } - else { - reader = new NetSocketReader(socket); - } - reader.setMessageFormat(messageFormat); - reader.setMaxMessageSize(receiveBufferSize); + NetSocketReader reader = SocketIoUtils.createNetReader(messageFormat, + customSocketReaderClass, socket, this.receiveBufferSize, + this.soReceiveBufferSize); while (true) { try { if (reader.assembleData()) { @@ -145,16 +131,16 @@ public class TcpNetReceivingChannelAdapter extends } /** - * @param customSocketReader the customSocketReader to set + * @param customSocketReaderClass the customSocketReader to set * @throws ClassNotFoundException */ @SuppressWarnings("unchecked") public void setCustomSocketReaderClassName( String customSocketReaderClassName) throws ClassNotFoundException { if (customSocketReaderClassName != null) { - this.customSocketReader = (Class) Class + this.customSocketReaderClass = (Class) Class .forName(customSocketReaderClassName); - if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReader))) { + if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) { throw new IllegalArgumentException("Custom socket reader must be of type NetSocketReader"); } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java index c7d2b280d4..2d1a6685ef 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java @@ -15,12 +15,11 @@ */ package org.springframework.integration.ip.tcp; -import java.lang.reflect.Constructor; import java.net.Socket; import javax.net.SocketFactory; -import org.springframework.beans.BeanUtils; +import org.springframework.integration.ip.util.SocketIoUtils; /** @@ -31,7 +30,7 @@ import org.springframework.beans.BeanUtils; public class TcpNetSendingMessageHandler extends AbstractTcpSendingMessageHandler { - protected Class customSocketWriter; + protected Class customSocketWriterClass; /** * Constructs a TcpNetSendingMessageHandler that sends data to the @@ -46,22 +45,22 @@ public class TcpNetSendingMessageHandler extends protected volatile Socket socket; /** - * if + * @return the socket + */ + protected Socket getSocket() { + return socket; + } + + /** * @return the writer */ protected synchronized SocketWriter getWriter() { - if (writer == null) { + if (this.writer == null) { try { this.socket = SocketFactory.getDefault().createSocket(this.host, this.port); this.setSocketAttributes(socket); - NetSocketWriter writer; - if (messageFormat == MessageFormats.FORMAT_CUSTOM){ - Constructor ctor = customSocketWriter.getConstructor(Socket.class); - writer = BeanUtils.instantiateClass(ctor, socket); - } else { - writer = new NetSocketWriter(socket); - } - writer.setMessageFormat(messageFormat); + NetSocketWriter writer = SocketIoUtils.createNetWriter(messageFormat, + customSocketWriterClass, socket); this.writer = writer; } catch (Exception e) { logger.error("Error creating SocketWriter", e); @@ -71,16 +70,16 @@ public class TcpNetSendingMessageHandler extends } /** - * @param customSocketWriter the customSocketWriter to set + * @param customSocketWriterClassName the customSocketWriterClassName to set * @throws ClassNotFoundException */ @SuppressWarnings("unchecked") public void setCustomSocketWriterClassName( String customSocketWriterClassName) throws ClassNotFoundException { if (customSocketWriterClassName != null) { - this.customSocketWriter = (Class) Class + this.customSocketWriterClass = (Class) Class .forName(customSocketWriterClassName); - if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriter))) { + if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) { throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter"); } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java index 6275b0f733..c9c829906f 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -17,7 +17,6 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; -import java.lang.reflect.Constructor; import java.net.InetSocketAddress; import java.net.Socket; import java.net.SocketException; @@ -29,8 +28,8 @@ import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; -import org.springframework.beans.BeanUtils; import org.springframework.integration.core.Message; +import org.springframework.integration.ip.util.SocketIoUtils; /** * Tcp Receiving Channel adapter that uses a {@link java.nio.channels.SocketChannel}. @@ -46,7 +45,7 @@ public class TcpNioReceivingChannelAdapter extends protected ServerSocketChannel serverChannel; protected boolean usingDirectBuffers; - protected Class customSocketReader; + protected Class customSocketReaderClass; /** * Constructs a TcpNioReceivingChannelAdapter to listen on the port. @@ -152,22 +151,10 @@ public class TcpNioReceivingChannelAdapter extends * @return The NioSocketReader. */ private NioSocketReader createSocketReader(final SelectionKey key) { - NioSocketReader reader = null; SocketChannel channel = (SocketChannel) key.channel(); - if (messageFormat == MessageFormats.FORMAT_CUSTOM) { - try { - Constructor ctor = customSocketReader - .getConstructor(SocketChannel.class); - reader = BeanUtils.instantiateClass(ctor, channel); - } catch (Exception e) { - logger.error("Error creating SocketReader", e); - } - } else { - reader = new NioSocketReader(channel); - } - reader.setUsingDirectBuffers(usingDirectBuffers); - reader.setMessageFormat(messageFormat); - reader.setMaxMessageSize(receiveBufferSize); + NioSocketReader reader = SocketIoUtils.createNioReader(messageFormat, + this.customSocketReaderClass, channel, this.receiveBufferSize, + this.receiveBufferSize, this.usingDirectBuffers); return reader; } @@ -212,15 +199,15 @@ public class TcpNioReceivingChannelAdapter extends } /** - * @param customSocketReader the customSocketReader to set + * @param customSocketReaderClassName the customSocketReaderClassName to set * @throws ClassNotFoundException */ @SuppressWarnings("unchecked") public void setCustomSocketReaderClassName(String customSocketReaderClassName) throws ClassNotFoundException { - this.customSocketReader = (Class) Class + this.customSocketReaderClass = (Class) Class .forName(customSocketReaderClassName); - if (!(NioSocketReader.class.isAssignableFrom(this.customSocketReader))) { + if (!(NioSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) { throw new IllegalArgumentException("Custom socket reader must be of type NioSocketReader"); } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java index b9576155db..9fca8d0c48 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java @@ -15,10 +15,9 @@ */ package org.springframework.integration.ip.tcp; -import java.lang.reflect.Constructor; import java.nio.channels.SocketChannel; -import org.springframework.beans.BeanUtils; +import org.springframework.integration.ip.util.SocketIoUtils; /** @@ -32,7 +31,7 @@ public class TcpNioSendingMessageHandler extends protected boolean usingDirectBuffers; - protected Class customSocketWriter; + protected Class customSocketWriterClass; protected int buffsPerConnection = 5; @@ -48,20 +47,13 @@ public class TcpNioSendingMessageHandler extends * @return the socket */ protected synchronized SocketWriter getWriter() { - if (socketChannel == null) { + if (this.socketChannel == null) { try { - socketChannel = SocketChannel.open(this.destinationAddress); + this.socketChannel = SocketChannel.open(this.destinationAddress); this.setSocketAttributes(socketChannel.socket()); - NioSocketWriter writer; - if (messageFormat == MessageFormats.FORMAT_CUSTOM){ - Constructor ctor = customSocketWriter - .getConstructor(SocketChannel.class, int.class, int.class); - writer = BeanUtils.instantiateClass(ctor, socketChannel, buffsPerConnection, soSendBufferSize); - } else { - writer = new NioSocketWriter(socketChannel, buffsPerConnection, soSendBufferSize); - } - writer.setMessageFormat(messageFormat); - writer.setUsingDirectBuffers(usingDirectBuffers); + NioSocketWriter writer = SocketIoUtils.createNioWriter(messageFormat, + customSocketWriterClass, socketChannel, + buffsPerConnection, soSendBufferSize, usingDirectBuffers); this.writer = writer; } catch (Exception e) { logger.error("Error creating SocketWriter", e); @@ -79,16 +71,16 @@ public class TcpNioSendingMessageHandler extends } /** - * @param customSocketWriter the customSocketWriter to set + * @param customSocketWriterClassName the customSocketWriterClassName to set * @throws ClassNotFoundException */ @SuppressWarnings("unchecked") public void setCustomSocketWriterClassName( String customSocketWriterClassName) throws ClassNotFoundException { if (customSocketWriterClassName != null) { - this.customSocketWriter = (Class) Class + this.customSocketWriterClass = (Class) Class .forName(customSocketWriterClassName); - if (!(NioSocketWriter.class.isAssignableFrom(this.customSocketWriter))) { + if (!(NioSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) { throw new IllegalArgumentException("Custom socket writer must be of type NioSocketWriter"); } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/util/SocketIoUtils.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/util/SocketIoUtils.java new file mode 100644 index 0000000000..e8cc695fd6 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/util/SocketIoUtils.java @@ -0,0 +1,162 @@ +/* + * 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.util; + +import java.lang.reflect.Constructor; +import java.net.Socket; +import java.nio.channels.SocketChannel; + +import org.springframework.beans.BeanUtils; +import org.springframework.integration.ip.tcp.MessageFormats; +import org.springframework.integration.ip.tcp.NetSocketReader; +import org.springframework.integration.ip.tcp.NetSocketWriter; +import org.springframework.integration.ip.tcp.NioSocketReader; +import org.springframework.integration.ip.tcp.NioSocketWriter; +import org.springframework.integration.message.MessageMappingException; + +/** + * @author Gary Russell + * + */ +public class SocketIoUtils { + + /** + * @param messageFormat + * @param customSocketReaderClass + * @param socket + * @param receiveBufferSize + * @param soReceiveBufferSize + * @return + */ + public static NetSocketReader createNetReader(int messageFormat, + Class customSocketReaderClass, + Socket socket, + int receiveBufferSize, + int soReceiveBufferSize) { + NetSocketReader reader; + if (messageFormat == MessageFormats.FORMAT_CUSTOM) { + try { + Constructor ctor = + customSocketReaderClass.getConstructor(Socket.class); + reader = BeanUtils.instantiateClass(ctor, socket); + if (soReceiveBufferSize > 0) { + socket.setReceiveBufferSize(soReceiveBufferSize); + } + } catch (Exception e) { + throw new MessageMappingException("Failed to instantiate custom reader", e); + } + } + else { + reader = new NetSocketReader(socket); + } + reader.setMessageFormat(messageFormat); + reader.setMaxMessageSize(receiveBufferSize); + return reader; + } + + /** + * @param messageFormat + * @param socket + * @param customSocketWriterClass + * @return + */ + public static NetSocketWriter createNetWriter(int messageFormat, + Class customSocketWriterClass, Socket socket) { + NetSocketWriter writer; + if (messageFormat == MessageFormats.FORMAT_CUSTOM){ + try { + Constructor ctor = customSocketWriterClass.getConstructor(Socket.class); + writer = BeanUtils.instantiateClass(ctor, socket); + } catch (Exception e) { + throw new MessageMappingException("Failed to instantiate custom writer", e); + } + } else { + writer = new NetSocketWriter(socket); + } + writer.setMessageFormat(messageFormat); + return writer; + } + + /** + * @param messageFormat + * @param customSocketReaderClass + * @param socket + * @param receiveBufferSize + * @param usingDirectBuffers + * @param soReceiveBufferSize + * @return + */ + public static NioSocketReader createNioReader(int messageFormat, + Class customSocketReaderClass, + SocketChannel channel, + int receiveBufferSize, + int soReceiveBufferSize, + boolean usingDirectBuffers ) { + NioSocketReader reader; + if (messageFormat == MessageFormats.FORMAT_CUSTOM) { + try { + Constructor ctor = + customSocketReaderClass.getConstructor(SocketChannel.class); + reader = BeanUtils.instantiateClass(ctor, channel); + if (soReceiveBufferSize > 0) { + channel.socket().setReceiveBufferSize(soReceiveBufferSize); + } + } catch (Exception e) { + throw new MessageMappingException("Failed to instantiate custom reader", e); + } + } + else { + reader = new NioSocketReader(channel); + } + reader.setMessageFormat(messageFormat); + reader.setMaxMessageSize(receiveBufferSize); + reader.setUsingDirectBuffers(usingDirectBuffers); + return reader; + } + + /** + * @param messageFormat + * @param socket + * @param customSocketWriter + * @param maxBuffers + * @param sendBufferSize + * @param usingDirectBuffers + * @return + */ + public static NioSocketWriter createNioWriter(int messageFormat, + Class customSocketWriterClass, + SocketChannel channel, + int maxBuffers, + int sendBufferSize, + boolean usingDirectBuffers) { + NioSocketWriter writer; + if (messageFormat == MessageFormats.FORMAT_CUSTOM){ + try { + Constructor ctor = customSocketWriterClass + .getConstructor(SocketChannel.class, int.class, int.class); + writer = BeanUtils.instantiateClass(ctor, channel, maxBuffers, sendBufferSize); + } catch (Exception e) { + throw new MessageMappingException("Failed to instantiate custom writer", e); + } + } else { + writer = new NioSocketWriter(channel, maxBuffers, sendBufferSize); + } + writer.setMessageFormat(messageFormat); + writer.setUsingDirectBuffers(usingDirectBuffers); + return writer; + } + +} diff --git a/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 9824a6da69..3db1baf2f3 100644 --- a/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -79,14 +79,44 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp - Defines an inbound Gateway for receiving and replying to incoming IP packets. + Defines an inbound Gateway for receiving and replying to incoming tcp messages. - + + + +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. + + + + + + +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. + + + + + + + + + + + + Defines an outbound Gateway for sending and receiving responses over TCP. + + + + + + @@ -125,7 +155,7 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp - + @@ -133,6 +163,8 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp + + @@ -169,6 +201,7 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp + @@ -179,7 +212,6 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp - @@ -190,7 +222,6 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp - diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 88e9f191f6..84110f179b 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -189,5 +189,21 @@ so-timeout="126" /> + + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 340b6fba64..31b4544466 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -23,12 +23,15 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.ApplicationContext; import org.springframework.integration.ip.tcp.CustomNetSocketReader; import org.springframework.integration.ip.tcp.CustomNetSocketWriter; import org.springframework.integration.ip.tcp.CustomNioSocketReader; import org.springframework.integration.ip.tcp.CustomNioSocketWriter; import org.springframework.integration.ip.tcp.MessageFormats; import org.springframework.integration.ip.tcp.SimpleTcpNetInboundGateway; +import org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway; +import org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGatewayTests; import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler; import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter; @@ -50,6 +53,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) public class ParserUnitTests { + @Autowired + ApplicationContext ctx; + @Autowired @Qualifier(value="testInUdp") UnicastReceivingChannelAdapter udpIn; @@ -92,6 +98,10 @@ public class ParserUnitTests { @Autowired SimpleTcpNetInboundGateway simpleTcpNetInboundGateway; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.SimpleTcpNetOutboundGateway#0") + SimpleTcpNetOutboundGateway simpleTcpNetOutboundGateway; @Test public void testInUdp() { @@ -120,7 +130,7 @@ public class ParserUnitTests { public void testInTcpNio() { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInNio); assertTrue(tcpInNio.getPort() >= 5200); - assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReader")); + assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReaderClass")); assertEquals(false, dfa.getPropertyValue("usingDirectBuffers")); assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat")); assertEquals(27, dfa.getPropertyValue("poolSize")); @@ -134,7 +144,7 @@ public class ParserUnitTests { public void testInTcpNioDirect() { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInNioDirect); assertTrue(tcpInNioDirect.getPort() >= 5300); - assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReader")); + assertEquals(CustomNioSocketReader.class, dfa.getPropertyValue("customSocketReaderClass")); assertEquals(true, dfa.getPropertyValue("usingDirectBuffers")); assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat")); assertEquals(27, dfa.getPropertyValue("poolSize")); @@ -148,7 +158,7 @@ public class ParserUnitTests { public void testInTcpNet() { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpInNet); assertTrue(tcpInNet.getPort() >= 5400); - assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReader")); + assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReaderClass")); assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat")); assertEquals(27, dfa.getPropertyValue("poolSize")); assertEquals(true, dfa.getPropertyValue("soKeepAlive")); @@ -201,7 +211,7 @@ public class ParserUnitTests { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutNio); assertTrue(tcpOutNio.getPort() >= 6200); assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat")); - assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriter")); + assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass")); assertEquals(true, dfa.getPropertyValue("soKeepAlive")); assertEquals(3, dfa.getPropertyValue("soLinger")); assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); @@ -217,7 +227,7 @@ public class ParserUnitTests { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutNioDirect); assertTrue(tcpOutNioDirect.getPort() >= 6300); assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat")); - assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriter")); + assertEquals(CustomNioSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass")); assertEquals(true, dfa.getPropertyValue("soKeepAlive")); assertEquals(3, dfa.getPropertyValue("soLinger")); assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); @@ -233,7 +243,7 @@ public class ParserUnitTests { DirectFieldAccessor dfa = new DirectFieldAccessor(tcpOutNet); assertTrue(tcpOutNet.getPort() >= 6400); assertEquals(MessageFormats.FORMAT_STX_ETX, dfa.getPropertyValue("messageFormat")); - assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriter")); + assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass")); assertEquals(true, dfa.getPropertyValue("soKeepAlive")); assertEquals(3, dfa.getPropertyValue("soLinger")); assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); @@ -251,8 +261,8 @@ public class ParserUnitTests { TcpNetReceivingChannelAdapter delegate = (TcpNetReceivingChannelAdapter) dfa .getPropertyValue("delegate"); DirectFieldAccessor delegateDfa = new DirectFieldAccessor(delegate); - assertEquals(CustomNetSocketReader.class, delegateDfa.getPropertyValue("customSocketReader")); - assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriter")); + assertEquals(CustomNetSocketReader.class, delegateDfa.getPropertyValue("customSocketReaderClass")); + assertEquals(CustomNetSocketWriter.class, dfa.getPropertyValue("customSocketWriterClass")); assertEquals(true, dfa.getPropertyValue("soKeepAlive")); assertEquals(123, dfa.getPropertyValue("receiveBufferSize")); assertEquals(124, dfa.getPropertyValue("soReceiveBufferSize")); @@ -260,5 +270,28 @@ public class ParserUnitTests { assertEquals(126, dfa.getPropertyValue("soTimeout")); assertEquals(23, dfa.getPropertyValue("poolSize")); - } + } + + @Test + public void testOutGateway() { + DirectFieldAccessor dfa = new DirectFieldAccessor(simpleTcpNetOutboundGateway); + assertTrue(simpleTcpNetOutboundGateway.getPort() >= 6500); + assertEquals(MessageFormats.FORMAT_CRLF, dfa.getPropertyValue("messageFormat")); + TcpNetSendingMessageHandler handler = (TcpNetSendingMessageHandler) dfa + .getPropertyValue("handler"); + DirectFieldAccessor delegateDfa = new DirectFieldAccessor(handler); + assertEquals(CustomNetSocketReader.class, dfa.getPropertyValue("customSocketReaderClass")); + assertEquals(CustomNetSocketWriter.class, delegateDfa.getPropertyValue("customSocketWriterClass")); + assertEquals(true, delegateDfa.getPropertyValue("soKeepAlive")); + assertEquals(224, dfa.getPropertyValue("soReceiveBufferSize")); + assertEquals(225, delegateDfa.getPropertyValue("soSendBufferSize")); + assertEquals(226, delegateDfa.getPropertyValue("soTimeout")); + } + + @Test + public void test() { + String[] beanDefinitionNames = ctx.getBeanDefinitionNames(); + for (String x : beanDefinitionNames) + System.out.println(x); + } } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests-context.xml new file mode 100644 index 0000000000..38480d8c7a --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests-context.xml @@ -0,0 +1,18 @@ + + + + + + + + + + + + + + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java new file mode 100644 index 0000000000..ae24ff2da8 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGatewayTests.java @@ -0,0 +1,94 @@ +/* + * 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 org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import static org.junit.Assert.assertEquals; + +/** + * @author Gary Russell + * + */ +@ContextConfiguration(locations="SimpleTcpNetInboundGatewayTests-context.xml") +@RunWith(SpringJUnit4ClassRunner.class) +public class SimpleTcpNetOutboundGatewayTests { + + @Autowired + @Qualifier("gatewayCrLf") + private SimpleTcpNetInboundGateway inboundGatewayCrLf; + + @Autowired + @Qualifier("gatewayStxEtx") + private SimpleTcpNetInboundGateway inboundGatewayStxEtx; + + @Autowired + @Qualifier("gatewayLength") + private SimpleTcpNetInboundGateway inboundGatewayLength; + + @Autowired + @Qualifier("gatewayCustom") + private SimpleTcpNetInboundGateway inboundGatewayCustom; + + @Test + public void testOutboundCrLf() throws Exception { + SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway + ("localhost", inboundGatewayCrLf.getPort()); + gateway.setMessageFormat(MessageFormats.FORMAT_CRLF); + Message message = MessageBuilder.withPayload("test").build(); + byte[] bytes = (byte[]) gateway.handleRequestMessage(message); + assertEquals("echo:test", new String(bytes)); + } + + @Test + public void testOutboundStxEtx() throws Exception { + SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway + ("localhost", inboundGatewayStxEtx.getPort()); + gateway.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + Message message = MessageBuilder.withPayload("test").build(); + byte[] bytes = (byte[]) gateway.handleRequestMessage(message); + assertEquals("echo:test", new String(bytes)); + } + + @Test + public void testOutboundLength() throws Exception { + SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway + ("localhost", inboundGatewayLength.getPort()); + gateway.setMessageFormat(MessageFormats.FORMAT_LENGTH_HEADER); + Message message = MessageBuilder.withPayload("test").build(); + byte[] bytes = (byte[]) gateway.handleRequestMessage(message); + assertEquals("echo:test", new String(bytes)); + } + + @Test + public void testOutboundCustom() throws Exception { + SimpleTcpNetOutboundGateway gateway = new SimpleTcpNetOutboundGateway + ("localhost", inboundGatewayCustom.getPort()); + gateway.setMessageFormat(MessageFormats.FORMAT_CUSTOM); + gateway.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNetSocketReader"); + gateway.setCustomSocketWriterClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter"); + Message message = MessageBuilder.withPayload("test").build(); + byte[] bytes = (byte[]) gateway.handleRequestMessage(message); + assertEquals("echo:test", new String(bytes).trim()); + } +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java index 5576fb81e1..1c5be5c980 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java @@ -104,7 +104,7 @@ public class UdpUnicastEndToEndTests implements Runnable { // tell the receiver to we're done doneProcessing.countDown(); } - assertTrue(firstReceived.await(2, TimeUnit.SECONDS)); + assertTrue(firstReceived.await(5, TimeUnit.SECONDS)); assertEquals(testingIpText, new String(finalMessage.getPayload())); }