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

INT-1279 Namespace Updates For New Gateways; Remove Old Gateways/Adapters; Migrate All Tests to New Gateways/Adapters

Browse Source
This commit is contained in:
Gary Russell
2010-08-13 22:05:59 +00:00
parent 64624490bf
commit a54c7ed5f1
78 changed files with 1827 additions and 6784 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

@@ -19,9 +19,7 @@ package org.springframework.integration.ip.config;
import org.springframework.beans.factory.BeanCreationException;
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;
import org.w3c.dom.Element;
@@ -33,8 +31,6 @@ import org.w3c.dom.Element;
*/
public abstract class IpAdapterParserUtils {
static final String IP_PROTOCOL_ATTRIBUTE = "protocol";
static final String UDP_MULTICAST = "multicast";
static final String MULTICAST_ADDRESS = "multicast-address";
@@ -87,8 +83,6 @@ public abstract class IpAdapterParserUtils {
static final String SO_TRAFFIC_CLASS = "so-traffic-class";
static final String CLOSE = "close";
static final String LOCAL_ADDRESS = "local-address";
static final String TASK_EXECUTOR = "task-executor";
@@ -104,6 +98,13 @@ public abstract class IpAdapterParserUtils {
static final String TCP_CONNECTION_FACTORY = "connection-factory";
public static final String INTERCEPTOR_FACTORY_CHAIN = "interceptor-factory-chain";
public static final String REQUEST_TIMEOUT = "request-timeout";
public static final String REPLY_TIMEOUT = "reply-timeout";
public static final String REPLY_CHANNEL = "reply-channel";
/**
* Adds a constructor-arg to the provided bean definition builder
@@ -151,26 +152,6 @@ public abstract class IpAdapterParserUtils {
builder.addConstructorArgValue(port);
}
/**
* Asserts that a protocol attribute (udp or tcp) is supplied,
* @param element
* @param parserContext
* @return The value of the attribute.
* @throws BeanCreationException if attribute not provided or invalid.
*/
static String getProtocol(Element element, ParserContext parserContext) {
String protocol = element.getAttribute(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE);
if (!StringUtils.hasText(protocol)) {
parserContext.getReaderContext().error(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE +
" is required for an IP channel adapter", element);
}
if (!protocol.equals("tcp") && !protocol.equals("udp")) {
parserContext.getReaderContext().error(IpAdapterParserUtils.IP_PROTOCOL_ATTRIBUTE +
" must be 'tcp' or 'udp' for an IP channel adapter", element);
}
return protocol;
}
/**
* Asserts that a port attribute is supplied.
* @param element
@@ -200,22 +181,6 @@ public abstract class IpAdapterParserUtils {
return multicast;
}
/**
* Sets the close attribute, if present.
* @param element
*/
static void setClose(Element element, BeanDefinitionBuilder builder) {
String close = element.getAttribute(IpAdapterParserUtils.CLOSE);
if (!StringUtils.hasText(close)) {
close = "false";
}
if (close.equals("true")) {
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.CLOSE),
close);
}
}
/**
* Gets the use-nio attribute, if present; if not returns 'false'.
* @param element
@@ -229,58 +194,6 @@ public abstract class IpAdapterParserUtils {
return useNio;
}
/**
* Gets the message-format attribute, if present; if not returns
* {@link MessageFormats#FORMAT_LENGTH_HEADER}.
* @param element
* @return The value of the attribute or false.
*/
static Integer getMessageFormat(Element element) {
String messageFormat = element.getAttribute(IpAdapterParserUtils.MESSAGE_FORMAT);
if (!StringUtils.hasText(messageFormat)) {
return MessageFormats.FORMAT_LENGTH_HEADER;
}
if (messageFormat.equals("length-header")) {
return MessageFormats.FORMAT_LENGTH_HEADER;
}
if (messageFormat.equals("stx-etx")) {
return MessageFormats.FORMAT_STX_ETX;
}
if (messageFormat.equals("crlf")) {
return MessageFormats.FORMAT_CRLF;
}
if (messageFormat.equals("serialized")) {
return MessageFormats.FORMAT_JAVA_SERIALIZED;
}
if (messageFormat.equals("custom")) {
return MessageFormats.FORMAT_CUSTOM;
}
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).

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java
View File

@@ -27,10 +27,10 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
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());
this.registerBeanDefinitionParser("outbound-gateway", new IpOutboundGatewayParser());
this.registerBeanDefinitionParser("udp-inbound-channel-adapter", new UdpInboundChannelAdapterParser());
this.registerBeanDefinitionParser("udp-outbound-channel-adapter", new UdpOutboundChannelAdapterParser());
this.registerBeanDefinitionParser("tcp-inbound-gateway", new TcpInboundGatewayParser());
this.registerBeanDefinitionParser("tcp-outbound-gateway", new TcpOutboundGatewayParser());
this.registerBeanDefinitionParser("tcp-connection-factory", new TcpConnectionParser());
this.registerBeanDefinitionParser("tcp-inbound-channel-adapter", new TcpInboundChannelAdapterParser());
this.registerBeanDefinitionParser("tcp-outbound-channel-adapter", new TcpOutboundChannelAdapterParser());

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java
View File

@@ -67,7 +67,7 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser {
}
} else {
parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE +
" must be 'client' or 'server' for an IP channel adapter", element);
" must be 'client' or 'server' for a TCP Connection Factory", element);
}
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,

20
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundGatewayParser.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundGatewayParser.java
View File

@@ -16,40 +16,32 @@
package org.springframework.integration.ip.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.core.Conventions;
import org.springframework.integration.config.xml.AbstractInboundGatewayParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.SimpleTcpNetInboundGateway;
import org.springframework.integration.ip.tcp.TcpInboundGateway;
import org.w3c.dom.Element;
/**
* @author Gary Russell
* @since 2.0
*/
public class IpInboundGatewayParser extends AbstractInboundGatewayParser {
public class TcpInboundGatewayParser extends AbstractInboundGatewayParser {
@Override
protected Class<?> getBeanClass(Element element) {
return SimpleTcpNetInboundGateway.class;
return TcpInboundGateway.class;
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !attributeName.equals(IpAdapterParserUtils.MESSAGE_FORMAT)
&& !attributeName.equals(IpAdapterParserUtils.TASK_EXECUTOR)
return !attributeName.equals(IpAdapterParserUtils.TCP_CONNECTION_FACTORY)
&& super.isEligibleAttribute(attributeName);
}
@Override
protected void doPostProcess(BeanDefinitionBuilder builder, Element element) {
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TASK_EXECUTOR);
}
IpAdapterParserUtils.TCP_CONNECTION_FACTORY); }
}

17
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundGatewayParser.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundGatewayParser.java
View File

@@ -28,7 +28,7 @@ import org.w3c.dom.Element;
* @author Gary Russell
* @since 2.0
*/
public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser {
public class TcpOutboundGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected String getInputChannelAttributeName() {
@@ -38,14 +38,15 @@ public class IpOutboundGatewayParser extends AbstractConsumerEndpointParser {
@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);
"org.springframework.integration.ip.tcp.TcpOutboundGateway");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.TCP_CONNECTION_FACTORY);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
IpAdapterParserUtils.REPLY_CHANNEL);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
IpAdapterParserUtils.setClose(element, builder);
IpAdapterParserUtils.REQUEST_TIMEOUT);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.REPLY_TIMEOUT);
return builder;
}

46
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpInboundChannelAdapterParser.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpInboundChannelAdapterParser.java
View File

@@ -16,19 +16,15 @@
package org.springframework.integration.ip.config;
import org.w3c.dom.Element;
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.AbstractChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter;
import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Channel Adapter that receives UDP datagram packets and maps them to Messages.
@@ -36,16 +32,10 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @since 2.0
*/
public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser {
public class UdpInboundChannelAdapterParser extends AbstractChannelAdapterParser {
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
String protocol = IpAdapterParserUtils.getProtocol(element, parserContext);
BeanDefinitionBuilder builder = null;
if (protocol.equals("tcp")) {
builder = parseTcp(element, parserContext);
} else if (protocol.equals("udp")) {
builder = parseUdp(element, parserContext);
}
BeanDefinitionBuilder builder = parseUdp(element, parserContext);
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);
@@ -100,34 +90,4 @@ public class IpInboundChannelAdapterParser extends AbstractChannelAdapterParser
return builder;
}
/**
* @param element
* @param parserContext
* @return
*/
private BeanDefinitionBuilder parseTcp(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder;
String useNio = IpAdapterParserUtils.getUseNio(element);
if (useNio.equals("false")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNetReceivingChannelAdapter.class);
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNioReceivingChannelAdapter.class);
}
addPortToConstructor(element, builder, parserContext);
builder.addPropertyValue(
Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT),
IpAdapterParserUtils.getMessageFormat(element));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.CUSTOM_SOCKET_READER_CLASS_NAME);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.USING_DIRECT_BUFFERS);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
IpAdapterParserUtils.SO_KEEP_ALIVE);
IpAdapterParserUtils.setClose(element, builder);
return builder;
}
}

34
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java → spring-integration-ip/src/main/java/org/springframework/integration/ip/config/UdpOutboundChannelAdapterParser.java
View File

@@ -21,8 +21,6 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler;
import org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler;
import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
import org.springframework.util.StringUtils;
@@ -32,17 +30,10 @@ import org.w3c.dom.Element;
* @author Gary Russell
* @since 2.0
*/
public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
public class UdpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
String protocol = IpAdapterParserUtils.getProtocol(element, parserContext);
BeanDefinitionBuilder builder = null;
if (protocol.equals("tcp")) {
builder = parseTcp(element, parserContext);
}
else if (protocol.equals("udp")) {
builder = parseUdp(element, parserContext);
}
BeanDefinitionBuilder builder = parseUdp(element, parserContext);
IpAdapterParserUtils.addCommonSocketOptions(builder, element);
return builder.getBeanDefinition();
}
@@ -103,25 +94,4 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt
return builder;
}
/**
* @param element
* @param parserContext
* @return
*/
private BeanDefinitionBuilder parseTcp(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder;
String useNio = IpAdapterParserUtils.getUseNio(element);
if (useNio.equals("false")) {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNetSendingMessageHandler.class);
}
else {
builder = BeanDefinitionBuilder
.genericBeanDefinition(TcpNioSendingMessageHandler.class);
}
IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);
IpAdapterParserUtils.addOutboundTcpAttributes(element, builder);
return builder;
}
}

154
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java
View File

@@ -1,154 +0,0 @@
/*
* 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.io.IOException;
/**
* Abstract SocketReader that handles data in 3 standard, and one custom
* format. The default format is {@link MessageFormats#FORMAT_LENGTH_HEADER} in which
* the message consists of a 4 byte integer (in network byte order) containing
* the length of data that follows. {@link MessageFormats#FORMAT_STX_ETX}
* indicates a message where the data begins with STX (0x02) and ends with
* ETX (0x03); the STX and ETX are not part of the data. {@link MessageFormats#FORMAT_CRLF}
* indicates a message followed by carriage return and line feed '\r\n'.
* FORMAT_LENGTH_HEADER can be used for {@link java.net.Socket} and
* {@link java.nio.channels.SocketChannel} implementations are provided for
* the standard formats. Users requiring other formats should subclass the
* appropriate implementation, and provide an implementation for
* {@link #assembleDataCustomFormat()} which is invoked by {@link #assembleData()}
* when the format is {@link MessageFormats#FORMAT_CUSTOM}.
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractSocketReader implements SocketReader, MessageFormats {
protected int messageFormat = FORMAT_LENGTH_HEADER;
/**
* The assembled data; must contain a reference when assembleData()
* returns true; will be set to null when getAssembledData() is called.
*/
protected Object assembledData;
protected int maxMessageSize = 1024 * 60;
/**
* Assembles data in format {@link #FORMAT_LENGTH_HEADER}.
* @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or
* < 0 if socket closed before any data for a message is received.
* @throws IOException
*/
protected abstract int assembleDataLengthFormat() throws IOException;
/**
* Assembles data in format {@link #FORMAT_STX_ETX}.
* @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or
* < 0 if socket closed before any data for a message is received.
* @throws IOException
*/
protected abstract int assembleDataStxEtxFormat() throws IOException;
/**
* Assembles data in format {@link #FORMAT_CRLF}.
* @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or
* < 0 if socket closed before any data for a message is received.
* @throws IOException
*/
protected abstract int assembleDataCrLfFormat() throws IOException;
/**
* Assembles data in format {@link #FORMAT_JAVA_SERIALIZED}
* @return SocketReader.MESSAGE_COMPLETE when message is assembled, otherwise SocketReader.MESSAGE_IMCOMPLETE, or
* < 0 if socket closed before any data for a message is received.
* @throws IOException
*/
protected abstract int assembleDataSerializedFormat() throws IOException;
/**
* Assembles data in format {@link #FORMAT_CUSTOM}. Implementations must
* return false until the message is completely assembled, at which time
* the implementation must update assembledData to reference the assembled
* message.
* @return True when a message is completely assembled.
* @throws IOException
*/
protected abstract int assembleDataCustomFormat() throws IOException;
public int assembleData() throws IOException {
int result;
try {
switch (this.messageFormat) {
case FORMAT_LENGTH_HEADER:
result = assembleDataLengthFormat();
break;
case FORMAT_STX_ETX:
result = assembleDataStxEtxFormat();
break;
case FORMAT_CRLF:
result = assembleDataCrLfFormat();
break;
case FORMAT_JAVA_SERIALIZED:
result = assembleDataSerializedFormat();
break;
case FORMAT_CUSTOM:
result = assembleDataCustomFormat();
break;
default:
throw new UnsupportedOperationException(
"Unsupported message format: " + messageFormat);
}
if (result < 0) {
doClose();
}
return result;
} catch (IOException e) {
doClose();
throw e;
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.SocketReader#getAssembledData()
*/
public Object getAssembledData() {
Object assembledData = this.assembledData;
this.assembledData = null;
if (assembledData instanceof byte[] &&
((byte[]) assembledData).length == 0) {
return null;
}
return assembledData;
}
/**
* Called after an exception; close the transport.
*/
protected abstract void doClose();
/**
* @param messageFormat the messageFormat to set,
*/
public void setMessageFormat(int messageFormat) {
this.messageFormat = messageFormat;
}
public void setMaxMessageSize(int maxMessageSize) {
this.maxMessageSize = maxMessageSize;
}
}

123
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java
View File

@@ -1,123 +0,0 @@
/*
* 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.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Abstract SocketWriter that handles data in 3 standard, and one custom
* format. The default format is {@link MessageFormats#FORMAT_LENGTH_HEADER} in which
* the message consists of a 4 byte integer (in network byte order) containing
* the length of data that follows. {@link MessageFormats#FORMAT_STX_ETX}
* indicates a message where the data begins with STX (0x02) and ends with
* ETX (0x03); the STX and ETX are not part of the data. {@link MessageFormats#FORMAT_CRLF}
* indicates a message followed by carriage return and line feed '\r\n'.
* FORMAT_LENGTH_HEADER can be used for {@link java.net.Socket} and
* {@link java.nio.channels.SocketChannel} implementations are provided for
* the standard formats. Users requiring other formats should subclass the
* appropriate implementation, and provide an implementation for
* {@link #writeCustomFormat(Object)} which is invoked by {@link #write(Object)}
* when the format is {@link MessageFormats#FORMAT_CUSTOM}.
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractSocketWriter implements SocketWriter, MessageFormats {
protected int messageFormat = FORMAT_LENGTH_HEADER;
protected final Log logger = LogFactory.getLog(this.getClass());
/*
* @see org.springframework.integration.ip.tcp.SocketWriter#write(Object)
*/
public synchronized void write(Object object) throws IOException {
try {
switch (this.messageFormat) {
case FORMAT_LENGTH_HEADER:
writeLengthFormat((byte[]) object);
return;
case FORMAT_STX_ETX:
writeStxEtxFormat((byte[]) object);
return;
case FORMAT_CRLF:
writeCrLfFormat((byte[]) object);
return;
case FORMAT_JAVA_SERIALIZED:
writeSerializedFormat(object);
return;
case FORMAT_CUSTOM:
writeCustomFormat(object);
return;
default:
throw new UnsupportedOperationException(
"Unsupported message format: " + messageFormat);
}
}
catch (IOException e) {
doClose();
throw e;
}
}
/**
* Called when an IO error
*/
protected abstract void doClose();
/**
* Write the length of the data in a 4 byte integer (in network byte
* order) before the data itself.
* @param bytes The bytes to write.
* @throws IOException
*/
protected abstract void writeLengthFormat(byte[] bytes) throws IOException;
/**
* Write an STX (0x02) followed by the data, followed by ETX (0x03).
* @param bytes The bytes to write.
*/
protected abstract void writeStxEtxFormat(byte[] bytes) throws IOException;
/**
* Write the data, followed by carriage return, line feed ('\r\n').
*/
protected abstract void writeCrLfFormat(byte[] bytes) throws IOException;
/**
* Write the data, followed by carriage return, line feed ('\r\n').
*/
protected abstract void writeSerializedFormat(Object object) throws IOException;
/**
* Write the data using some custom protocol.
*/
protected abstract void writeCustomFormat(Object object) throws IOException;
/**
* @param messageFormat the messageFormat to set
*/
public void setMessageFormat(int messageFormat) {
this.messageFormat = messageFormat;
}
}

107
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java
View File

@@ -1,107 +0,0 @@
/*
* 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.ip.AbstractInternetProtocolReceivingChannelAdapter;
/**
* Abstract class for tcp/ip incoming channel adapters. Implementations
* for {@link java.net.Socket} and {@link java.nio.channels.SocketChannel}
* are provided.
*
* @author Gary Russell
* @since 2.0
*
*/
public abstract class AbstractTcpReceivingChannelAdapter extends
AbstractInternetProtocolReceivingChannelAdapter {
protected volatile SocketMessageMapper mapper = new SocketMessageMapper();
protected volatile boolean soKeepAlive;
protected volatile int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
protected volatile boolean close;
/**
* Constructs a receiving channel adapter that listens on the port.
* @param port The port to listen on.
*/
public AbstractTcpReceivingChannelAdapter(int port) {
super(port);
}
/**
* Checks that we have a task executor and calls
* {@link #server()}.
*/
public void run() {
if (logger.isDebugEnabled()) {
logger.debug(this.getClass().getSimpleName() + " running on port: " + port);
}
checkTaskExecutor("TCP-Incoming-Msg-Handler");
server();
}
/**
* Establishes the server.
*/
protected abstract void server();
/**
* Sets soTimeout, soKeepAlive and tcpNoDelay according to the configured
* properties.
* @param socket The socket.
* @throws SocketException
*/
protected void setSocketOptions(Socket socket) throws SocketException {
socket.setSoTimeout(this.soTimeout);
if (this.soReceiveBufferSize > 0) {
socket.setReceiveBufferSize(this.soReceiveBufferSize);
}
socket.setKeepAlive(this.soKeepAlive);
}
/**
* @see Socket#setKeepAlive(boolean)
* @param soKeepAlive the soKeepAlive to set
*/
public void setSoKeepAlive(boolean soKeepAlive) {
this.soKeepAlive = soKeepAlive;
}
/**
* @see MessageFormats
* @param messageFormat the messageFormat to set
*/
public void setMessageFormat(int messageFormat) {
this.messageFormat = messageFormat;
mapper.setMessageFormat(messageFormat);
}
/**
* @param close the close to set
*/
public void setClose(boolean close) {
this.close = close;
}
}

174
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java
View File

@@ -1,174 +0,0 @@
/*
* 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.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import org.springframework.integration.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
import org.springframework.integration.mapping.MessageMappingException;
/**
* Abstract class for TCP sending message handlers. Implementations
* for {@link java.net.Socket} and {@link java.nio.channels.SocketChannel}
* are provided.
*
* @author Gary Russell
*
*/
public abstract class AbstractTcpSendingMessageHandler extends
AbstractInternetProtocolSendingMessageHandler {
protected SocketMessageMapper mapper = new SocketMessageMapper();
protected AbstractSocketWriter writer;
protected boolean soTcpNoDelay = false;
protected int soLinger = -1;
protected int soTrafficClass = -1;
protected boolean soKeepAlive = false;
protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
/**
* Constructs a message handler that sends messages to the specified
* host and port.
* @param host The host.
* @param port The port.
*/
public AbstractTcpSendingMessageHandler(String host, int port) {
super(host, port);
}
/**
* Sets socket attributes on the socket.
* @param socket The socket.
* @throws SocketException
*/
protected void setSocketAttributes(Socket socket) throws SocketException {
if (this.soTimeout >= 0) {
socket.setSoTimeout(this.soTimeout);
}
if (this.soSendBufferSize > 0) {
socket.setSendBufferSize(this.soSendBufferSize);
}
socket.setTcpNoDelay(this.soTcpNoDelay);
if (soLinger >= 0) {
socket.setSoLinger(true, this.soLinger);
}
if (soTrafficClass >= 0) {
socket.setTrafficClass(this.soTrafficClass);
}
socket.setKeepAlive(this.soKeepAlive);
}
/**
* Returns the socket writer after instantiating it, if necessary.
* @return The writer.
*/
protected abstract SocketWriter getWriter();
/**
* Writes the message payload to the underlying socket, using the specified
* message format.
* @see org.springframework.integration.core.MessageHandler#handleMessage(org.springframework.integration.Message)
*/
public void handleMessage(final Message<?> message) throws MessageRejectedException,
MessageHandlingException, MessageDeliveryException {
try {
doWrite(message);
} catch (MessageMappingException e) {
// retry - socket may have closed
if (e.getCause() instanceof IOException) {
doWrite(message);
} else {
throw e;
}
}
}
/**
* Method that actually does the write.
* @param message The message to write.
*/
protected void doWrite(Message<?> message) {
try {
Object object = mapper.fromMessage(message);
SocketWriter writer = this.getWriter();
if (writer == null) {
throw new MessageMappingException(message, "Failed to create SocketWriter");
}
writer.write(object);
} catch (Exception e) {
this.writer = null;
if (e instanceof MessageMappingException) {
throw (MessageMappingException) e;
}
throw new MessageMappingException(message, "Failed to map message", e);
}
}
/**
* @see Socket#setTcpNoDelay(boolean)
* @param soTcpNoDelay the soTcpNoDelay to set
*/
public void setSoTcpNoDelay(boolean soTcpNoDelay) {
this.soTcpNoDelay = soTcpNoDelay;
}
/**
* Enables SO_LINGER on the underlying socket.
* @see Socket#setSoLinger(boolean, int)
* @param soLinger the soLinger to set
*/
public void setSoLinger(int soLinger) {
this.soLinger = soLinger;
}
/**
* @see Socket#setTrafficClass(int)
* @param soTrafficClass the soTrafficClass to set
*/
public void setSoTrafficClass(int soTrafficClass) {
this.soTrafficClass = soTrafficClass;
}
/**
* @see Socket#setKeepAlive(boolean)
* @param soKeepAlive the soKeepAlive to set
*/
public void setSoKeepAlive(boolean soKeepAlive) {
this.soKeepAlive = soKeepAlive;
}
/**
* @see MessageFormats
* @param messageFormat the messageFormat to set
*/
public void setMessageFormat(int messageFormat) {
this.messageFormat = messageFormat;
mapper.setMessageFormat(messageFormat);
}
}

57
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java
View File

@@ -1,57 +0,0 @@
/*
* 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;
/**
* Definition of message formats supported by tcp channel adapters.
*
* @author Gary Russell
*
*/
public interface MessageFormats {
/**
* Message has format '&lt;length&gt;&lt;message&gt;'.
*/
public static final int FORMAT_LENGTH_HEADER = 1;
/**
* Message has format 'STX&lt;message&gt;ETX'.
*/
public static final int FORMAT_STX_ETX = 2;
/**
* Message has format '&lt;message&gt;\r\n'.
*/
public static final int FORMAT_CRLF = 3;
/**
* Not a real format; any formats less than this must use
* a byte[] payload (or a String for outbound). Formats higher
* have implied format (e.g. Serializable).
*/
public static final int FORMAT_IMPLICIT = 128;
/**
* Message contains a Serializable object
*/
public static final int FORMAT_JAVA_SERIALIZED = 129;
/**
* Message has custom format.
*/
public static final int FORMAT_CUSTOM = 255;
public static final int STX = 0x02;
public static final int ETX = 0x03;
}

228
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java
View File

@@ -1,228 +0,0 @@
/*
* 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.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.mapping.MessageMappingException;
/**
* A SocketReader that reads from a {@link java.net.Socket}. Threads
* calling {@link NetSocketReader#assembledData} will block until a message
* is completely assembled.
*
* @author Gary Russell
* @since 2.0
*
*/
public class NetSocketReader extends AbstractSocketReader {
protected final Log logger = LogFactory.getLog(getClass());
protected Socket socket;
protected ObjectInputStream objectInputStream;
/**
* Constructs a NetsocketReader which reads from the Socket.
* @param socket The socket.
*/
public NetSocketReader(Socket socket) {
this.socket = socket;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.SocketReader#read(java.nio.ByteBuffer)
*/
@Override
protected int assembleDataLengthFormat() throws IOException {
byte[] lengthPart = new byte[4];
int status = read(lengthPart, true);
if (status < 0) {
return status;
}
int messageLength = ByteBuffer.wrap(lengthPart).getInt();
if (logger.isDebugEnabled()) {
logger.debug("Message length is " + messageLength);
}
if (messageLength > this.maxMessageSize) {
throw new IOException("Message length " + messageLength +
" exceeds max message length: " + this.maxMessageSize);
}
byte[] messagePart = new byte[messageLength];
read(messagePart, false);
this.assembledData = messagePart;
return MESSAGE_COMPLETE;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat()
*/
@Override
protected int assembleDataStxEtxFormat() throws IOException {
InputStream inputStream = socket.getInputStream();
int bite = inputStream.read();
if (bite < 0) {
return bite;
}
if (bite != STX)
throw new MessageMappingException("Expected STX to begin message");
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
while ((bite = inputStream.read()) != ETX) {
checkClosure(bite);
buffer[n++] = (byte) bite;
if (n >= this.maxMessageSize) {
throw new IOException("ETX not found before max message length: "
+ this.maxMessageSize);
}
}
this.assembledData = new byte[n];
System.arraycopy(buffer, 0, this.assembledData, 0, n);
return MESSAGE_COMPLETE;
}
private void checkClosure(int bite) throws IOException {
if (bite < 0) {
logger.debug("Socket closed");
throw new IOException("Socket closed");
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat()
*/
@Override
protected int assembleDataCrLfFormat() throws IOException {
InputStream inputStream = socket.getInputStream();
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
int bite;
while (true) {
bite = inputStream.read();
if (bite < 0 && n == 0) {
return bite;
}
checkClosure(bite);
if (n > 0 && bite == '\n' && buffer[n-1] == '\r')
break;
buffer[n++] = (byte) bite;
if (n >= this.maxMessageSize) {
throw new IOException("CRLF not found before max message length: "
+ this.maxMessageSize);
}
};
this.assembledData = new byte[n-1];
System.arraycopy(buffer, 0, this.assembledData, 0, n-1);
return MESSAGE_COMPLETE;
}
@Override
protected int assembleDataSerializedFormat() throws IOException {
try {
if (this.objectInputStream == null) {
InputStream is = this.socket.getInputStream();
this.objectInputStream = new ObjectInputStream(is);
}
this.assembledData = this.objectInputStream.readObject();
} catch (EOFException ee) {
return SOCKET_CLOSED;
} catch (ClassNotFoundException e) {
throw new IOException(e.getMessage());
}
return SocketReader.MESSAGE_COMPLETE;
}
/**
* Throws {@link UnsupportedOperationException}; custom implementations can
* subclass this class and provide an implementation for this method.
* @throws IOException
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCustomFormat
*
*/
@Override
protected int assembleDataCustomFormat() throws IOException {
throw new UnsupportedOperationException("Need to subclass for this format");
}
/**
* Reads data from the socket and puts the data in buffer. Blocks until
* buffer is full or a socket timeout occurs.
* @param buffer
* @param header true if we are reading the header
* @return < 0 if socket closed and not in the middle of a message
* @throws IOException
*/
protected int read(byte[] buffer, boolean header) throws IOException {
int lengthRead = 0;
int needed = buffer.length;
while (lengthRead < needed) {
int len;
len = this.socket.getInputStream().read(buffer, lengthRead,
needed - lengthRead);
if (len < 0 && header && lengthRead == 0) {
return len;
}
if (len < 0)
logger.debug("socket closed after " + lengthRead + " of " + needed);
checkClosure(len);
lengthRead += len;
if (logger.isDebugEnabled()) {
logger.debug("Read " + len + " bytes, buffer is now at " +
lengthRead + " of " +
needed);
}
}
return 0;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#doClose()
*/
@Override
protected void doClose() {
try {
socket.close();
} catch (IOException e) {
logger.error("Error on close", e);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.SocketReader#getAddress()
*/
public InetAddress getAddress() {
return this.socket.getInetAddress();
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.SocketReader#getSocket()
*/
public Socket getSocket() {
return socket;
}
}

114
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java
View File

@@ -1,114 +0,0 @@
/*
* 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.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.nio.ByteBuffer;
/**
* A {@link SocketWriter} that writes to a {@link java.net.Socket}. The
* data is wrapped in a wire protocol based on the messageFormat property.
*
* @author Gary Russell
* @since 2.0
*/
public class NetSocketWriter extends AbstractSocketWriter {
protected Socket socket;
protected ObjectOutputStream objectOutputStream;
/**
* Constructs a NetSocketWriter for the Socket.
*
* @param socket The socket.
*/
public NetSocketWriter(Socket socket) {
this.socket = socket;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCrLfFormat(byte[])
*/
@Override
protected void writeCrLfFormat(byte[] bytes) throws IOException {
OutputStream outputStream = this.socket.getOutputStream();
outputStream.write(bytes);
outputStream.write('\r');
outputStream.write('\n');
outputStream.flush();
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[])
*/
@Override
protected void writeSerializedFormat(Object object) throws IOException {
if (this.objectOutputStream == null) {
OutputStream os = this.socket.getOutputStream();
this.objectOutputStream = new ObjectOutputStream(os);
}
this.objectOutputStream.writeObject(object);
this.objectOutputStream.flush();
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[])
*/
@Override
protected void writeCustomFormat(Object object) throws IOException {
throw new UnsupportedOperationException("Need to subclass for this format");
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeLengthFormat(byte[])
*/
@Override
protected void writeLengthFormat(byte[] bytes) throws IOException {
ByteBuffer lengthPart = ByteBuffer.allocate(4);
lengthPart.putInt(bytes.length);
OutputStream outputStream = this.socket.getOutputStream();
outputStream.write(lengthPart.array());
outputStream.write(bytes);
outputStream.flush();
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeStxEtxFormat(byte[])
*/
@Override
protected void writeStxEtxFormat(byte[] bytes) throws IOException {
OutputStream outputStream = this.socket.getOutputStream();
outputStream.write(STX);
outputStream.write(bytes);
outputStream.write(ETX);
outputStream.flush();
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#doClose()
*/
@Override
protected void doClose() {
try {
this.socket.close();
} catch (IOException e) {
logger.error("Error on close", e);
}
}
}

367
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java
View File

@@ -1,367 +0,0 @@
/*
* 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.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.mapping.MessageMappingException;
/**
* A non-blocking SocketReader that reads from a {@link java.nio.channels.SocketChannel}.
*
* @author Gary Russell
* @since 2.0
*
*/
public class NioSocketReader extends AbstractSocketReader {
protected final Log logger = LogFactory.getLog(getClass());
protected SocketChannel channel;
protected boolean usingDirectBuffers;
protected ByteBuffer lengthPart;
protected ByteBuffer dataPart;
protected ByteBuffer rawBuffer;
protected ByteBuffer buildBuffer;
protected boolean building;
/**
* Constructs an NioSocketReader which reads from the SocketChannel.
* @param channel The channel.
*/
public NioSocketReader(SocketChannel channel) {
this.channel = channel;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.SocketReader#assembleData()
*/
@Override
public int assembleDataLengthFormat() throws IOException {
if (lengthPart == null) {
lengthPart = allocate(4);
}
if (lengthPart.hasRemaining()) {
int status = readChannel(lengthPart);
if (status < 0) {
if (lengthPart.remaining() == 4) {
// not in the middle of a message, clean close
return status;
}
throw new IOException("Channel closed");
}
return MESSAGE_INCOMPLETE;
}
if (dataPart == null) {
lengthPart.flip();
int messageLength = lengthPart.getInt();
if (logger.isDebugEnabled()) {
logger.debug("Message length is " + messageLength);
}
if (messageLength > maxMessageSize) {
throw new IOException("Message length " + messageLength +
" exceeds max message length " + maxMessageSize);
}
dataPart = ByteBuffer.allocate(messageLength);
}
if (dataPart.hasRemaining()) {
int status = readChannel(dataPart);
if (status < 0) {
throw new IOException("Channel closed");
}
if (dataPart.hasRemaining()) {
return MESSAGE_INCOMPLETE;
}
}
assembledData = dataPart.array();
lengthPart = dataPart = null;
return MESSAGE_COMPLETE;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat()
*/
@Override
protected int assembleDataStxEtxFormat() throws IOException {
int len = readChannelNonDeterministic();
if (len > 0) {
byte bite = this.rawBuffer.get();
int count = 0;
if (!this.building) {
if (bite != STX) {
throw new MessageMappingException("Expected STX, received " + Integer.toHexString(bite));
}
this.building = true;
count++;
if (!this.rawBuffer.hasRemaining()) {
if (logger.isDebugEnabled()) {
logger.debug("Incomplete message, consumed 1 byte");
}
return MESSAGE_INCOMPLETE;
}
} else {
if (bite == ETX) {
finishAssembly();
return MESSAGE_COMPLETE;
}
this.buildBuffer.put(bite);
count++;
if (this.buildBuffer.position() >= this.buildBuffer.limit()) {
throw new IOException("ETX not found before max message length: "
+ maxMessageSize);
}
}
while (true) {
if (!this.rawBuffer.hasRemaining()) {
if (logger.isDebugEnabled()) {
logger.debug("Incomplete message, consumed " + count + " bytes");
}
return MESSAGE_INCOMPLETE;
}
bite = this.rawBuffer.get();
if (bite == ETX) {
break;
}
this.buildBuffer.put(bite);
count++;
if (this.buildBuffer.position() >= this.buildBuffer.limit()) {
throw new IOException("ETX not found before max message length: "
+ maxMessageSize);
}
}
if (logger.isDebugEnabled()) {
logger.debug("Consumed " + count + " bytes");
}
finishAssembly();
return MESSAGE_COMPLETE;
} else if (len == 0) {
logger.debug("Incomplete message, nothing to read");
return MESSAGE_INCOMPLETE;
} else {
logger.debug("Channel closed");
if (!this.building) {
// not in the middle of a message, clean close
return SOCKET_CLOSED;
}
this.building = false;
throw new IOException("Channel closed");
}
}
/**
*
*/
private void finishAssembly() {
byte[] assembledData = new byte[buildBuffer.position()];
System.arraycopy(this.buildBuffer.array(), 0, assembledData, 0, assembledData.length);
this.building = false;
this.buildBuffer.clear();
this.assembledData = assembledData;
logger.debug("Message assembly complete");
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat()
*/
@Override
protected int assembleDataCrLfFormat() throws IOException {
int len = readChannelNonDeterministic();
if (len > 0) {
int count = 0;
while (true) {
if (!rawBuffer.hasRemaining()) {
if (logger.isDebugEnabled()) {
logger.debug("Incomplete message, consumed " + count + " bytes");
}
return MESSAGE_INCOMPLETE;
}
byte bite = rawBuffer.get();
this.building = true;
if (bite == '\n' && this.buildBuffer.position() > 0) {
this.buildBuffer.position(this.buildBuffer.position() - 1);
if (this.buildBuffer.get() == '\r') {
this.buildBuffer.position(this.buildBuffer.position() - 1);
break;
}
}
this.buildBuffer.put(bite);
count++;
if (this.buildBuffer.position() >= this.buildBuffer.limit()) {
throw new IOException("CRLF not found before max message length: "
+ this.maxMessageSize);
}
}
if (logger.isDebugEnabled()) {
logger.debug("Consumed " + count + " bytes");
}
finishAssembly();
return MESSAGE_COMPLETE;
} else if (len == 0) {
logger.debug("Incomplete message, nothing to read");
return MESSAGE_INCOMPLETE;
} else {
logger.debug("Channel closed");
if (!this.building) {
// not in the middle of a message, clean close
return SOCKET_CLOSED;
}
this.building = false;
throw new IOException("Channel closed");
}
}
/**
* Throws {@link UnsupportedOperationException}; Java serialization is currently only
* supported using the NetSocketReader.
* @throws IOException
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCustomFormat
*
*/
protected int assembleDataSerializedFormat() throws IOException {
throw new UnsupportedOperationException("Serializable not supported using NIO");
}
/**
* Throws {@link UnsupportedOperationException}; custom implementations can
* subclass this class and provide an implementation.
* @throws IOException
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCustomFormat
*
*/
@Override
protected int assembleDataCustomFormat() throws IOException {
throw new UnsupportedOperationException("Need to subclass for this format");
}
/**
* Reads from the channel into the buffer. Reads as much data as is
* currently available in the channel.
* @param buffer
* @throws IOException
*/
protected int readChannel(ByteBuffer buffer) throws IOException {
try {
int len = channel.read(buffer);
if (len < 0) {
logger.debug("Socket closed");
return len;
}
if (logger.isDebugEnabled()) {
logger.debug("Read " + len + " bytes, buffer is now at " +
buffer.position() + " of " +
buffer.capacity());
}
return len;
} catch (IOException e) {
throw e;
}
}
/**
* Reads data into the rawBuffer for non-deterministic algorithms.
* @return bytes remaining in raw buffer or < 0 if channel closed
* @throws IOException
*/
protected int readChannelNonDeterministic() throws IOException {
if (rawBuffer == null) {
rawBuffer = allocate(maxMessageSize);
buildBuffer = ByteBuffer.allocate(maxMessageSize);
} else if (rawBuffer.hasRemaining()) {
if (logger.isDebugEnabled()) {
logger.debug("Raw buffer has " + rawBuffer.remaining() + " remaining");
}
return rawBuffer.remaining();
}
rawBuffer.clear();
int len = channel.read(rawBuffer);
if (len < 0) {
return len;
}
rawBuffer.flip();
if (logger.isDebugEnabled()) {
logger.debug("Read " + rawBuffer.limit() + " into raw buffer");
}
return rawBuffer.remaining();
}
/**
* Allocates a ByteBuffer of the requested length using normal or
* direct buffers, depending on the usingDirectBuffers field.
*/
protected ByteBuffer allocate(int length) {
ByteBuffer buffer;
if (usingDirectBuffers) {
buffer = ByteBuffer.allocateDirect(length);
} else {
buffer = ByteBuffer.allocate(length);
}
return buffer;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketReader#doClose()
*/
@Override
protected void doClose() {
try {
channel.close();
} catch (IOException e) {
logger.error("Error on close", e);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.SocketReader#getAddress()
*/
public InetAddress getAddress() {
return this.channel.socket().getInetAddress();
}
public boolean isUsingDirectBuffers() {
return usingDirectBuffers;
}
/**
* @param usingDirectBuffers the usingDirectBuffers to set
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.SocketReader#getSocket()
*/
public Socket getSocket() {
return channel.socket();
}
}

268
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java
View File

@@ -1,268 +0,0 @@
/*
* Copyright 2002-20/10 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.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
* A {@link SocketWriter} that writes to a {@link SocketChannel}. The
* data is wrapped in a wire protocol based on the messageFormat property.
*
* @author Gary Russell
* @since 2.0
*/
public class NioSocketWriter extends AbstractSocketWriter {
protected SocketChannel channel;
/**
* If true, direct buffers are used.
* @see ByteBuffer for more information
*/
protected boolean usingDirectBuffers;
/**
* A buffer containing the length part when the messageFormat is
* {@link MessageFormats#FORMAT_LENGTH_HEADER}.
*/
protected ByteBuffer lengthPart;
/**
* A buffer containing the STX for when the messageFormat is
* {@link MessageFormats#FORMAT_STX_ETX}.
*/
protected ByteBuffer stxPart;
/**
* A buffer containing the ETX for when the messageFormat is
* {@link MessageFormats#FORMAT_STX_ETX}.
*/
protected ByteBuffer etxPart;
/**
* A buffer containing the CRLF for when the messageFormat is
* {@link MessageFormats#FORMAT_CRLF}.
*/
protected ByteBuffer crLfPart;
/**
* If we are using direct buffers, we don't want to churn them using
* normal heap management. But,
* because we can have multiple threads writing and we might write in
* chunks, we need a dedicated buffer for each thread; up to a limit.
* We handle this with a blocking queue.
*/
protected BlockingQueue<ByteBuffer> buffers;
protected int maxBuffers = 2;
protected int bufferCount = 0;
private int sendBufferSize;
public NioSocketWriter(SocketChannel channel,
int maxBuffers,
int sendBufferSize) {
this.channel = channel;
this.maxBuffers = maxBuffers;
if (sendBufferSize <= 0) {
sendBufferSize = 2048;
}
this.sendBufferSize = sendBufferSize;
buffers = new LinkedBlockingQueue<ByteBuffer>(maxBuffers);
}
/**
* @param usingDirectBuffers whether direct buffers are to be used
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
protected ByteBuffer getBuffer() throws InterruptedException {
ByteBuffer buffer = this.buffers.poll();
if (buffer != null) {
buffer.clear();
return buffer;
}
synchronized (buffers) {
if (this.bufferCount < this.maxBuffers) {
bufferCount++;
return ByteBuffer.allocateDirect(this.sendBufferSize);
}
}
buffer = this.buffers.take();
buffer.clear();
return buffer;
}
protected void returnBuffer(ByteBuffer buffer) {
if (buffer != null) {
this.buffers.offer(buffer);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCrLfFormat(byte[])
*/
@Override
protected void writeCrLfFormat(byte[] bytes) throws IOException {
ByteBuffer buffer = null;
if (this.usingDirectBuffers) {
try {
checkBufferSize(bytes, 2);
buffer = getBuffer();
buffer.put(bytes);
buffer.put((byte) '\r');
buffer.put((byte) '\n');
buffer.flip();
this.channel.write(buffer);
return;
} catch (InterruptedException e) {
throw new IOException("Could not get buffer; interrupted");
} finally {
returnBuffer(buffer);
}
}
synchronized (channel) {
if (this.crLfPart == null) {
this.crLfPart = ByteBuffer.allocate(2);
this.crLfPart.put((byte) '\r');
this.crLfPart.put((byte) '\n');
}
this.channel.write(ByteBuffer.wrap(bytes));
this.crLfPart.flip();
this.channel.write(this.crLfPart);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[])
*/
@Override
protected void writeSerializedFormat(Object object) throws IOException {
throw new UnsupportedOperationException("Serializable not supported using NIO");
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[])
*/
@Override
protected void writeCustomFormat(Object object) throws IOException {
throw new UnsupportedOperationException("Need to subclass for this format");
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeLengthFormat(byte[])
*/
@Override
protected void writeLengthFormat(byte[] bytes) throws IOException {
ByteBuffer buffer = null;
if (this.usingDirectBuffers) {
try {
checkBufferSize(bytes, 4);
buffer = getBuffer();
buffer.putInt(bytes.length);
buffer.put(bytes);
buffer.flip();
this.channel.write(buffer);
return;
} catch (InterruptedException e) {
throw new IOException("Could not get buffer; interrupted");
} finally {
returnBuffer(buffer);
}
}
synchronized (channel) {
if (this.lengthPart == null) {
this.lengthPart = ByteBuffer.allocate(4);
} else {
this.lengthPart.clear();
}
this.lengthPart.putInt(bytes.length);
this.lengthPart.flip();
this.channel.write(this.lengthPart);
this.channel.write(ByteBuffer.wrap(bytes));
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeStxEtxFormat(byte[])
*/
@Override
protected void writeStxEtxFormat(byte[] bytes) throws IOException {
ByteBuffer buffer = null;
if (this.usingDirectBuffers) {
try {
checkBufferSize(bytes, 2);
buffer = getBuffer();
buffer.put((byte) STX);
buffer.put(bytes);
buffer.put((byte) ETX);
buffer.flip();
this.channel.write(buffer);
return;
} catch (InterruptedException e) {
throw new IOException("Could not get buffer; interrupted");
} finally {
returnBuffer(buffer);
}
}
synchronized (channel) {
if (this.stxPart == null) {
this.stxPart = ByteBuffer.allocate(1);
this.stxPart.put((byte) STX);
this.etxPart = ByteBuffer.allocate(1);
this.etxPart.put((byte) ETX);
}
this.stxPart.flip();
this.channel.write(this.stxPart);
this.channel.write(ByteBuffer.wrap(bytes));
this.etxPart.flip();
this.channel.write(this.etxPart);
}
}
/**
* @param bytes
* @throws IOException
*/
private void checkBufferSize(byte[] bytes, int pad) throws IOException {
if (bytes.length + pad > this.sendBufferSize) {
throw new IOException("Send buffer too small (" + sendBufferSize +
") increase so-send-buffer-size to at least " +
bytes.length + pad);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#doClose()
*/
@Override
protected void doClose() {
try {
this.channel.close();
} catch (IOException e) {
logger.error("Error on close", e);
}
}
}

266
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetInboundGateway.java
View File

@@ -1,266 +0,0 @@
/*
* 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.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.Executor;
import org.springframework.integration.Message;
import org.springframework.integration.gateway.AbstractMessagingGateway;
import org.springframework.integration.mapping.MessageMappingException;
/**
* 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
* @since 2.0
*
*/
public class SimpleTcpNetInboundGateway extends AbstractMessagingGateway {
protected SocketMessageMapper mapper = new SocketMessageMapper();
protected WriteCapableTcpNetReceivingChannelAdapter delegate;
protected int port;
protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
protected int poolSize = 2;
protected int receiveBufferSize = 2048;
protected boolean soKeepAlive;
protected int soReceiveBufferSize = -1;
protected int soSendBufferSize = -1;
protected int soTimeout = 0;
protected String customSocketReaderClassName;
protected Class<NetSocketWriter> customSocketWriterClass;
protected boolean close;
protected String localAddress;
protected Executor taskExecutor;
@Override
protected void doStart() {
super.doStart();
this.delegate.start();
}
@Override
protected void doStop() {
super.doStop();
this.delegate.stop();
}
@Override
protected void onInit() throws Exception {
this.delegate = new WriteCapableTcpNetReceivingChannelAdapter(this.port);
this.delegate.setMessageFormat(this.messageFormat);
this.delegate.setPoolSize(this.poolSize);
this.delegate.setReceiveBufferSize(this.receiveBufferSize);
this.delegate.setSoKeepAlive(this.soKeepAlive);
this.delegate.setSoReceiveBufferSize(this.soReceiveBufferSize);
this.delegate.setSoSendBufferSize(this.soSendBufferSize);
this.delegate.setSoTimeout(this.soTimeout);
this.delegate.setTaskScheduler(getTaskScheduler());
this.delegate.setCustomSocketReaderClassName(this.customSocketReaderClassName);
this.delegate.setClose(this.close);
this.delegate.setLocalAddress(this.localAddress);
this.delegate.setTaskExecutor(this.taskExecutor);
super.onInit();
}
@Override
protected Object fromMessage(Message<?> message) {
throw new MessageMappingException("Cannot map a message to an object in this gateway");
}
@Override
protected Message<?> toMessage(Object object) {
try {
return this.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;
mapper.setMessageFormat(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 this.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
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
if (customSocketWriterClassName != null) {
this.customSocketWriterClass = (Class<NetSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) {
throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter");
}
}
}
/**
* @param close the close to set
*/
public void setClose(boolean close) {
this.close = close;
}
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public boolean isListening() {
return delegate.isListening();
}
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
private class WriteCapableTcpNetReceivingChannelAdapter extends TcpNetReceivingChannelAdapter {
/**
* @param port
*/
public WriteCapableTcpNetReceivingChannelAdapter(int port) {
super(port);
}
@Override
protected void processMessage(NetSocketReader reader) {
Socket socket = reader.getSocket();
Message<?> message = sendAndReceiveMessage(reader);
NetSocketWriter writer = SocketIoUtils.createNetWriter(this.messageFormat,
customSocketWriterClass, socket);
try {
writer.write(this.mapper.fromMessage(message));
if (close) {
try {
socket.close();
} catch (IOException ioe) {
logger.error("Error on close", ioe);
}
}
} catch (Exception e) {
throw new MessageMappingException("Failed to map and send response", e);
}
}
@Override
protected void doStart() {
; super.doStart();
}
@Override
protected void setSocketOptions(Socket socket) throws SocketException {
super.setSocketOptions(socket);
if (soSendBufferSize > 0) {
socket.setSendBufferSize(soSendBufferSize);
}
}
}
}

226
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SimpleTcpNetOutboundGateway.java
View File

@@ -1,226 +0,0 @@
/*
* 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.io.IOException;
import java.net.Socket;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
/**
* 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<NetSocketReader> customSocketReaderClass;
protected int messageFormat;
protected int maxMessageSize = 2048;
protected int soReceiveBufferSize = -1;
protected NetSocketReader reader;
protected boolean close;
/**
* 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);
Socket socket = this.handler.getSocket();
if (this.reader == null ||
this.reader.getSocket() != socket) { // might have re-opened on error
this.reader = SocketIoUtils.createNetReader(this.messageFormat,
this.customSocketReaderClass, socket, this.maxMessageSize,
this.soReceiveBufferSize);
}
try {
if (this.reader.assembleData() < 0) {
throw new IOException("Socket closed");
}
Object object = this.reader.getAssembledData();
if (close) {
logger.debug("Closing socket because close=true");
this.handler.close();
}
return object;
} catch (Exception e) {
this.reader = null;
this.handler.close();
throw new MessagingException(requestMessage, e);
}
}
/**
* @see java.lang.Object#equals(Object)
* @return whether the MessageHandler delegate for this Gateway is equal to the provided object
*/
public boolean equals(Object obj) {
return handler.equals(obj);
}
/**
* @see org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler#getPort()
* @return the port number of the MessageHandler delegate for this Gateway
*/
public int getPort() {
return handler.getPort();
}
/**
* @see java.lang.Object#hashCode()
* @return hashcode value of the MessageHandler delegate for this Gateway
*/
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 customSocketReaderClassName the {@link NetSocketReader} class to use
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReaderClassName(
String customSocketReaderClassName) throws ClassNotFoundException {
if (customSocketReaderClassName != null) {
this.customSocketReaderClass = (Class<NetSocketReader>) 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);
}
/**
* @param close the close to set
*/
public void setClose(boolean close) {
this.close = close;
}
}

99
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketIoUtils.java
View File

@@ -15,13 +15,8 @@
*/
package org.springframework.integration.ip.tcp;
import java.lang.reflect.Constructor;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.channels.SocketChannel;
import org.springframework.beans.BeanUtils;
import org.springframework.integration.mapping.MessageMappingException;
/**
* @author Gary Russell
@@ -29,100 +24,6 @@ import org.springframework.integration.mapping.MessageMappingException;
*/
public class SocketIoUtils {
public static NetSocketReader createNetReader(int messageFormat,
Class<NetSocketReader> customSocketReaderClass,
Socket socket,
int receiveBufferSize,
int soReceiveBufferSize) {
NetSocketReader reader;
if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
try {
Constructor<NetSocketReader> 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;
}
public static NetSocketWriter createNetWriter(int messageFormat,
Class<NetSocketWriter> customSocketWriterClass, Socket socket) {
NetSocketWriter writer;
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
try {
Constructor<NetSocketWriter> 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;
}
public static NioSocketReader createNioReader(int messageFormat,
Class<NioSocketReader> customSocketReaderClass,
SocketChannel channel,
int receiveBufferSize,
int soReceiveBufferSize,
boolean usingDirectBuffers ) {
NioSocketReader reader;
if (messageFormat == MessageFormats.FORMAT_CUSTOM) {
try {
Constructor<NioSocketReader> 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;
}
public static NioSocketWriter createNioWriter(int messageFormat,
Class<NioSocketWriter> customSocketWriterClass,
SocketChannel channel,
int maxBuffers,
int sendBufferSize,
boolean usingDirectBuffers) {
NioSocketWriter writer;
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
try {
Constructor<NioSocketWriter> 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;
}
public static String getSocketId(Socket socket) {
InetAddress inetAddress = socket.getInetAddress();
String hostName = "";

120
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java
View File

@@ -1,120 +0,0 @@
/*
* 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.io.IOException;
import java.io.UnsupportedEncodingException;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.core.MessageBuilder;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
/**
* Maps incoming data from a {@link SocketReader} to a {@link Message} and from
* a Message to outgoing data forwarded to a {@link SocketWriter}.
* @author Gary Russell
*
*/
public class SocketMessageMapper implements
InboundMessageMapper<SocketReader>,
OutboundMessageMapper<Object> {
private volatile String charset = "UTF-8";
private volatile int messageFormat;
/* (non-Javadoc)
* @see org.springframework.integration.message.InboundMessageMapper#toMessage(java.lang.Object)
*/
public Message<Object> toMessage(SocketReader socketReader) throws Exception {
return fromRaw(socketReader);
}
/**
* Calls {@link SocketReader#getAssembledData()} and creates a message with
* the socket data (excluding any protocol parts) as the payload. The source
* hostname and ip address are added to the message headers.
* @param socketReader
* @return
* @throws IOException
*/
private Message<Object> fromRaw(SocketReader socketReader) throws IOException {
Object payload = socketReader.getAssembledData();
Message<Object> message = null;
if (payload != null) {
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;
}
/* (non-Javadoc)
* @see org.springframework.integration.message.OutboundMessageMapper#fromMessage(org.springframework.integration.core.Message)
*/
public Object fromMessage(Message<?> message) throws Exception {
if (this.messageFormat < MessageFormats.FORMAT_IMPLICIT) {
return getPayloadAsBytes(message);
}
return message.getPayload();
}
/**
* Extracts the payload as a byte array.
* @param message
* @return
*/
private byte[] getPayloadAsBytes(Message<?> message) {
byte[] bytes = null;
Object payload = message.getPayload();
if (payload instanceof byte[]) {
bytes = (byte[]) payload;
}
else if (payload instanceof String) {
try {
bytes = ((String) payload).getBytes(this.charset);
}
catch (UnsupportedEncodingException e) {
throw new MessageHandlingException(message, e);
}
}
else {
throw new MessageHandlingException(message, "The socket mapper expects " +
"either a byte array or String payload, but received: " + payload.getClass());
}
return bytes;
}
/**
* @param charset the charset to set
*/
public void setCharset(String charset) {
this.charset = charset;
}
public void setMessageFormat(int messageFormat) {
this.messageFormat = messageFormat;
}
}

67
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java
View File

@@ -1,67 +0,0 @@
/*
* 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.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
/**
* General interface for assembling message data from a TCP/IP Socket.
* Implementations for {@link java.net.Socket} and {@link java.nio.channels.SocketChannel}
* are provided.
*
* @author Gary Russell
* @since 2.0
*
*/
public interface SocketReader {
public static int SOCKET_CLOSED = -1;
public static int MESSAGE_INCOMPLETE = 0;
public static int MESSAGE_COMPLETE = 1;
/**
* Reads the data the socket and assembles
* packets of data into a complete message, depending on the format of that
* data.
* @return MESSAGE_COMPLETE when message is assembled, otherwise MESSAGE_IMCOMPLETE, or
* < 0 if socket closed before any data for a message is received.
* @throws IOException
*/
public int assembleData() throws IOException;
/**
* Retrieves the assembled tcp data or null if the data is not
* yet assembled. Once this method is called, the assembled data is
* again null until a new assembly is completed.
* @return The assembled data or null.
*/
public Object getAssembledData();
/**
* Returns the InetAddress of the underlying socket.
* @return The InetAddress.
*/
public InetAddress getAddress();
/**
* @return the Socket
*/
public Socket getSocket();
}

43
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java
View File

@@ -1,43 +0,0 @@
/*
* 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.io.IOException;
/**
* A general interface for writing to sockets.
*
* @author Gary Russell
*
*/
public interface SocketWriter {
/**
* Write the entire buffer to the underlying socket. Appropriate wire
* protocols will be implemented so the receiving side can decode and
* reassemble the message, if packetized by the network.
* @param object The object to write.
* @throws IOException
*/
void write(Object object) throws IOException;
/**
* @param messageFormat the messageFormat to set
*/
public void setMessageFormat(int messageFormat);
}

167
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java
View File

@@ -1,167 +0,0 @@
/*
* 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.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import javax.net.ServerSocketFactory;
import org.springframework.integration.Message;
/**
* Tcp Receiving Channel adapter that uses a {@link Socket}. Each
* connected socket uses a dedicated thread so the pool size must be set
* accordingly.
*
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNetReceivingChannelAdapter extends
AbstractTcpReceivingChannelAdapter {
protected ServerSocket serverSocket;
protected Class<NetSocketReader> customSocketReaderClass;
/**
* Constructs a TcpNetReceivingChannelAdapter that listens on the provided port.
* @param port the port on which to listen
*/
public TcpNetReceivingChannelAdapter(int port) {
super(port);
}
/**
* Creates the server socket, listens for incoming connections and schedules
* execution of the {@link #handleSocket(Socket)} method for each new
* connection.
*
* @see org.springframework.integration.ip.tcp.AbstractTcpReceivingChannelAdapter#server()
*/
@Override
protected void server() {
while (this.active) {
try {
if (this.localAddress == null) {
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(this.port, Math.abs(this.poolSize));
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(port, Math.abs(poolSize), whichNic);
}
this.listening = true;
while (true) {
final Socket socket = serverSocket.accept();
setSocketOptions(socket);
this.taskExecutor.execute(new Runnable() {
public void run() {
handleSocket(socket);
}});
}
} catch (IOException e) {
if (this.serverSocket != null) {
try {
this.serverSocket.close();
} catch (IOException e1) {}
}
this.listening = false;
this.serverSocket = null;
if (this.active) {
logger.error("Error on ServerSocket", e);
}
}
}
}
/**
* Constructs a {@link NetSocketReader} and calls its {@link NetSocketReader#assembledData}
* method repeatedly; for each assembled message, calls {@link #sendMessage(Message)} with
* the mapped message.
*/
protected void handleSocket(Socket socket) {
NetSocketReader reader = SocketIoUtils.createNetReader(messageFormat,
customSocketReaderClass, socket, this.receiveBufferSize,
this.soReceiveBufferSize);
while (true) {
try {
int messageStatus = reader.assembleData();
if (messageStatus < 0) {
return;
}
if (messageStatus == SocketReader.MESSAGE_COMPLETE) {
if (close) {
logger.debug("Closing socket because close=true");
try {
reader.getSocket().close();
} catch (IOException ioe) {
logger.error("Error on close", ioe);
}
}
processMessage(reader);
if (close) {
break;
}
}
} catch (Exception e) {
logger.error("processMessage failed", e);
return;
}
}
}
protected void processMessage(NetSocketReader reader)
throws Exception {
Message<Object> message = mapper.toMessage(reader);
if (message != null) {
sendMessage(message);
}
}
@Override
protected void doStop() {
super.doStop();
try {
this.serverSocket.close();
}
catch (Exception e) {
// ignore
}
}
/**
* @param customSocketReaderClassName the {@link NetSocketReader} class to use
* @throws ClassNotFoundException if the named class cannot be loaded
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReaderClassName(
String customSocketReaderClassName) throws ClassNotFoundException {
if (customSocketReaderClassName != null) {
this.customSocketReaderClass = (Class<NetSocketReader>) Class
.forName(customSocketReaderClassName);
if (!(NetSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) {
throw new IllegalArgumentException("Custom socket reader must be of type NetSocketReader");
}
}
}
}

101
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java
View File

@@ -1,101 +0,0 @@
/*
* 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 javax.net.SocketFactory;
/**
* TCP Sending Channel Adapter that that uses a {@link java.net.Socket}.
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNetSendingMessageHandler extends
AbstractTcpSendingMessageHandler {
protected Class<NetSocketWriter> customSocketWriterClass;
/**
* Constructs a TcpNetSendingMessageHandler that sends data to the
* specified host and port.
* @param host The host.
* @param port The port.
*/
public TcpNetSendingMessageHandler(String host, int port) {
super(host, port);
}
protected volatile Socket socket;
/**
* @return the socket
*/
protected Socket getSocket() {
return socket;
}
/**
* @return the writer
*/
protected synchronized SocketWriter getWriter() {
if (this.writer == null) {
try {
logger.debug("Opening new socket connection");
this.socket = SocketFactory.getDefault().createSocket(this.host, this.port);
this.setSocketAttributes(socket);
NetSocketWriter writer = SocketIoUtils.createNetWriter(messageFormat,
customSocketWriterClass, socket);
this.writer = writer;
} catch (Exception e) {
logger.error("Error creating SocketWriter", e);
}
}
return this.writer;
}
/**
* @param customSocketWriterClassName the customSocketWriterClassName to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
if (customSocketWriterClassName != null) {
this.customSocketWriterClass = (Class<NetSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NetSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) {
throw new IllegalArgumentException("Custom socket writer must be of type NetSocketWriter");
}
}
}
/**
* Close the underlying socket and prepare to establish a new socket on
* the next write.
*/
protected void close() {
this.writer.doClose();
this.writer = null;
}
public void setLocalAddress(String localAddress) {
logger.warn("localAddress not used on tcp outbound endpoints");
}
}

237
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java
View File

@@ -1,237 +0,0 @@
/*
* 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.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import org.springframework.integration.Message;
/**
* Tcp Receiving Channel adapter that uses a {@link java.nio.channels.SocketChannel}.
* Sockets are multiplexed across the pooled threads. More than one thread will
* be required with large numbers of connections and incoming traffic. The
* number of threads is controlled by the poolSize property.
*
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNioReceivingChannelAdapter extends
AbstractTcpReceivingChannelAdapter {
protected ServerSocketChannel serverChannel;
protected boolean usingDirectBuffers;
protected Class<NioSocketReader> customSocketReaderClass;
/**
* Constructs a TcpNioReceivingChannelAdapter to listen on the port.
* @param port The port.
*/
public TcpNioReceivingChannelAdapter(int port) {
super(port);
}
/**
* Opens a non-blocking {@link ServerSocketChannel}, registers it with a
* {@link Selector} and calls {@link #doSelect(ServerSocketChannel, Selector)}.
*
* @see org.springframework.integration.ip.tcp.AbstractTcpReceivingChannelAdapter#server()
*/
@Override
protected void server() {
try {
this.serverChannel = ServerSocketChannel.open();
this.listening = true;
this.serverChannel.configureBlocking(false);
if (this.localAddress == null) {
this.serverChannel.socket().bind(new InetSocketAddress(this.port),
Math.abs(this.poolSize));
} else {
InetAddress whichNic = InetAddress.getByName(this.localAddress);
this.serverChannel.socket().bind(new InetSocketAddress(whichNic, this.port),
Math.abs(this.poolSize));
}
final Selector selector = Selector.open();
this.serverChannel.register(selector, SelectionKey.OP_ACCEPT);
doSelect(this.serverChannel, selector);
} catch (IOException e) {
try {
serverChannel.close();
} catch (IOException e1) { }
this.listening = false;
this.serverChannel = null;
if (this.active) {
logger.error("Error on ServerSocketChannel", e);
}
}
}
/**
* Listens for incoming connections and for notifications that a connected
* socket is ready for reading.
* Accepts incoming connections, registers the new socket with the
* selector for reading.
* When a socket is ready for reading, unregisters the read interest and
* schedules a call to doRead which reads all available data. When the read
* is complete, the socket is again registered for read interest.
* @param server
* @param selector
* @throws IOException
* @throws ClosedChannelException
* @throws SocketException
*/
private void doSelect(ServerSocketChannel server, final Selector selector)
throws IOException, ClosedChannelException, SocketException {
while (active) {
int selectionCount = selector.select();
if (logger.isDebugEnabled())
logger.debug("Port " + port + " SelectionCount: " + selectionCount);
if (selectionCount > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
SocketChannel channel = null;
while (iterator.hasNext()) {
final SelectionKey key = iterator.next();
iterator.remove();
if (key.isAcceptable()) {
channel = server.accept();
channel.configureBlocking(false);
channel.register(selector, SelectionKey.OP_READ);
Socket socket = channel.socket();
setSocketOptions(socket);
}
else if (key.isReadable()) {
key.interestOps(key.interestOps() - key.readyOps());
if (key.attachment() == null) {
NioSocketReader reader = createSocketReader(key);
if (reader == null) {
continue;
}
key.attach(reader);
}
this.taskExecutor.execute(new Runnable() {
public void run() {
doRead(key);
if (key.channel().isOpen()) {
key.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
}});
}
else {
logger.error("Unexpected key: " + key);
}
}
}
}
}
/**
* Creates an NioSocketReader, either directly,or
* from the supplied class if {@link MessageFormats#FORMAT_CUSTOM}
* is used.
* @param key The selection key.
* @return The NioSocketReader.
*/
private NioSocketReader createSocketReader(final SelectionKey key) {
SocketChannel channel = (SocketChannel) key.channel();
NioSocketReader reader = SocketIoUtils.createNioReader(messageFormat,
this.customSocketReaderClass, channel, this.receiveBufferSize,
this.receiveBufferSize, this.usingDirectBuffers);
return reader;
}
/**
* Obtains the {@link NetSocketReader} associated with the channel
* and calls its {@link NetSocketReader#assembledData}
* method; if a message is fully assembled, calls {@link #sendMessage(Message)} with the
* mapped message.
*/
private void doRead(SelectionKey key) {
NioSocketReader reader = (NioSocketReader) key.attachment();
try {
int messageStatus = reader.assembleData();
if (messageStatus < 0) {
return;
}
if (messageStatus == SocketReader.MESSAGE_COMPLETE) {
if (close) {
logger.debug("Closing channel because close=true");
try {
key.channel().close();
} catch (IOException ioe) {
logger.error("Error on close", ioe);
}
}
Message<Object> message;
message = mapper.toMessage(reader);
if (message != null) {
sendMessage(message);
}
}
} catch (Exception e) {
logger.error("Failure on read or message send", e);
}
}
@Override
protected void doStop() {
super.doStop();
try {
this.serverChannel.close();
}
catch (Exception e) {
// ignore
}
}
/**
* @param usingDirectBuffers Set true if you wish to use direct buffers
* for NIO operations.
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
/**
* @param customSocketReaderClassName the customSocketReaderClassName to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketReaderClassName(String customSocketReaderClassName)
throws ClassNotFoundException {
this.customSocketReaderClass = (Class<NioSocketReader>) Class
.forName(customSocketReaderClassName);
if (!(NioSocketReader.class.isAssignableFrom(this.customSocketReaderClass))) {
throw new IllegalArgumentException("Custom socket reader must be of type NioSocketReader");
}
}
}

107
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java
View File

@@ -1,107 +0,0 @@
/*
* 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.nio.channels.SocketChannel;
/**
* TCP Sending Channel Adapter that that uses a {@link java.nio.channels.SocketChannel}.
*
* @author Gary Russell
* @since 2.0
*
*/
public class TcpNioSendingMessageHandler extends
AbstractTcpSendingMessageHandler {
protected volatile SocketChannel socketChannel;
protected boolean usingDirectBuffers;
protected Class<NioSocketWriter> customSocketWriterClass;
protected int buffsPerConnection = 5;
/**
* @param host
* @param port
*/
public TcpNioSendingMessageHandler(String host, int port) {
super(host, port);
}
/**
* @return the socket
*/
protected synchronized SocketWriter getWriter() {
if (this.socketChannel == null) {
try {
logger.debug("Creating new SocketChannel");
this.socketChannel = SocketChannel.open(this.destinationAddress);
this.setSocketAttributes(socketChannel.socket());
NioSocketWriter writer = SocketIoUtils.createNioWriter(messageFormat,
customSocketWriterClass, socketChannel,
buffsPerConnection, soSendBufferSize, usingDirectBuffers);
this.writer = writer;
} catch (Exception e) {
logger.error("Error creating SocketWriter", e);
}
}
return this.writer;
}
/**
* @param usingDirectBuffers Set true if you wish to use direct buffers
* for NIO operations.
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
/**
* @param customSocketWriterClassName the customSocketWriterClassName to set
* @throws ClassNotFoundException
*/
@SuppressWarnings("unchecked")
public void setCustomSocketWriterClassName(
String customSocketWriterClassName) throws ClassNotFoundException {
if (customSocketWriterClassName != null) {
this.customSocketWriterClass = (Class<NioSocketWriter>) Class
.forName(customSocketWriterClassName);
if (!(NioSocketWriter.class.isAssignableFrom(this.customSocketWriterClass))) {
throw new IllegalArgumentException("Custom socket writer must be of type NioSocketWriter");
}
}
}
/**
* If direct buffers are being used, sets the max number of
* buffers allowed per connection. Defaults to 5. It is unlikely
* this would ever need to be changed. Each buffer is set at the
* soSendBufferSize or, if not set, 2048 bytes.
*
* @param buffsPerConnection the buffsPerConnection to set
*/
public void setBuffsPerConnection(int buffsPerConnection) {
this.buffsPerConnection = buffsPerConnection;
}
public void setLocalAddress(String localAddress) {
logger.warn("localAddress not used on tcp outbound endpoints");
}
}

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpOutboundGateway.java
View File

@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
import org.springframework.integration.Message;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
@@ -93,6 +94,7 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
logger.error("Tcp Gateway exception", e);
throw new MessagingException("Failed to send or receive", e);
} finally {
if (haveSemaphore) {
@@ -180,4 +182,11 @@ public class TcpOutboundGateway extends AbstractReplyProducingMessageHandler
}
}
/**
* 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);
}
}

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

@@ -40,8 +40,6 @@ public class TcpReceivingChannelAdapter
protected ServerSocket serverSocket;
protected Class<NetSocketReader> customSocketReaderClass;
protected ConnectionFactory clientConnectionFactory;
protected ConnectionFactory serverConnectionFactory;

17
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -25,7 +25,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.commons.serializer.InputStreamingConverter;
import org.springframework.commons.serializer.OutputStreamingConverter;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter;
import org.springframework.util.Assert;
@@ -37,7 +37,7 @@ import org.springframework.util.Assert;
*
*/
public abstract class AbstractConnectionFactory
implements ConnectionFactory, Runnable, Lifecycle {
implements ConnectionFactory, Runnable, SmartLifecycle {
protected Log logger = LogFactory.getLog(this.getClass());
@@ -341,6 +341,17 @@ public abstract class AbstractConnectionFactory
return connection;
}
public int getPhase() {
return 0;
}
public boolean isAutoStartup() {
return true;
}
public void stop(Runnable callback) {
stop();
callback.run();
}
}

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java
View File

@@ -20,6 +20,7 @@ import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import javax.net.ServerSocketFactory;
@@ -50,6 +51,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
* I/O errors on the server socket/channel are logged and the factory is stopped.
*/
public void run() {
ServerSocket theServerSocket = null;
if (this.listener == null) {
logger.info("No listener bound to server connection factory; will not read; exiting...");
return;
@@ -63,6 +65,7 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(port, Math.abs(poolSize), whichNic);
}
theServerSocket = this.serverSocket;
this.listening = true;
logger.info("Listening on port " + this.port);
while (true) {
@@ -76,7 +79,10 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto
}
} catch (Exception e) {
this.listening = false;
if (this.active) {
// don't log an error if we had a good socket once and now it's closed
if (e instanceof SocketException && theServerSocket != null) {
logger.warn("Server Socket closed");
} else if (this.active) {
logger.error("Error on ServerSocket", e);
}
this.active = false;

1
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java
View File

@@ -46,6 +46,7 @@ public class TcpNioClientConnectionFactory extends
protected Map<SocketChannel, TcpNioConnection> connections = new ConcurrentHashMap<SocketChannel, TcpNioConnection>();
protected BlockingQueue<SocketChannel> newChannels = new LinkedBlockingQueue<SocketChannel>();
/**
* Creates a TcpNioClientConnectionFactory for connections to the host and port.

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

@@ -27,6 +27,7 @@ import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.SocketIoUtils;
@@ -60,6 +61,8 @@ public class TcpNioConnection extends AbstractTcpConnection {
private boolean active = true;
private long lastRead;
private AtomicInteger executionControl = new AtomicInteger();
/**
* Constructs a TcpNetConnection for the SocketChannel.
@@ -151,15 +154,20 @@ public class TcpNioConnection extends AbstractTcpConnection {
logger.debug("TcpListener exiting - no listener and not single use");
return;
}
if (active) {
while (active) {
try {
while (pipedInputStream.available() > 0) {
while (this.socketChannel.isOpen() &&
this.pipedInputStream.available() > 0) {
convertAndSend();
}
} catch (IOException e) {
logger.error("Unexpected exception, exiting...", e);
return;
}
// currently no more work to do
if (this.executionControl.decrementAndGet() < 0) {
break;
}
}
} finally {
logger.debug("Nio message assembler exiting...");
@@ -167,7 +175,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
private synchronized void convertAndSend() throws IOException {
if (this.pipedInputStream.available() <= 0) {
if (!this.socketChannel.isOpen() || this.pipedInputStream.available() <= 0) {
return;
}
Message<?> message = null;
@@ -236,10 +244,12 @@ public class TcpNioConnection extends AbstractTcpConnection {
pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());
pipedOutputStream.flush();
if (!socketChannel.isBlocking()) {
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newSingleThreadExecutor();
}
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newSingleThreadExecutor();
}
if (this.executionControl.incrementAndGet() <= 1) {
// only execute run() if we don't already have one running
this.executionControl.set(1);
this.taskExecutor.execute(this);
}
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java
View File

@@ -70,6 +70,7 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
try {
this.serverChannel = ServerSocketChannel.open();
this.listening = true;
logger.info("Listening on port " + this.port);
this.serverChannel.configureBlocking(false);
if (this.localAddress == null) {
this.serverChannel.socket().bind(new InetSocketAddress(this.port),
@@ -144,11 +145,13 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
logger.debug("Selection key no longer valid");
}
else if (key.isAcceptable()) {
logger.debug("New accept");
channel = server.accept();
channel.configureBlocking(false);
Socket socket = channel.socket();
setSocketAttributes(socket);
TcpNioConnection connection = createTcpNioConnection(channel);
connection.setTaskExecutor(this.taskExecutor);
connection.setLastRead(now);
connections.put(channel, connection);
channel.register(selector, SelectionKey.OP_READ, connection);

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

@@ -19,15 +19,15 @@
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="inbound-channel-adapter">
<xsd:element name="udp-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an inbound Channel Adapter for receiving incoming IP packets.
Defines a udp inbound Channel Adapter for receiving incoming udp packets.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="ipAdapterType">
<xsd:extension base="udpAdapterType">
<xsd:attribute name="pool-size" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
@@ -39,23 +39,6 @@ its configuration specifies the number of threads.
</xsd:attribute>
<xsd:attribute name="receive-buffer-size" type="xsd:string" />
<xsd:attribute name="multicast-address" 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="close" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation>
If true, the socket will be closed after a message is received and sent to the
outbound channel. If false, the socket will remain open ready to receive the
next message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -70,15 +53,15 @@ next message.
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:element name="udp-outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an outbound IP packet-sending Channel Adapter.
Defines an outbound UDP packet-sending Channel Adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="ipAdapterType">
<xsd:extension base="udpAdapterType">
<xsd:attribute name="host" type="xsd:string" />
<xsd:attribute name="acknowledge" type="xsd:string" />
<xsd:attribute name="ack-host" type="xsd:string" />
@@ -86,17 +69,6 @@ next message.
<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: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" />
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -111,115 +83,6 @@ task executors such as a WorkManagerTaskExecutor.
</xsd:complexType>
</xsd:element>
<xsd:element name="inbound-gateway">
<xsd:annotation>
<xsd:documentation>
Defines an inbound Gateway for receiving and replying to incoming tcp messages.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="gatewayType">
<xsd:attribute name="pool-size" type="xsd:string" >
<xsd:annotation>
<xsd:documentation>
The number of threads that will be used for socket/channel handling. Only applies
if an external task-executor is NOT being used. When using an external task executor,
its configuration specifies the number of threads.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specifies a specific Executor to be used for socket handling. If not supplied, an internal
pooled executor will be used (See pool-size). Needed on some platforms that require the use of specific
task executors such as a WorkManagerTaskExecutor.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-gateway">
<xsd:annotation>
<xsd:documentation>
Defines an outbound Gateway for sending and receiving responses over TCP.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="gatewayType">
<xsd:attribute name="host" 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:element name="tcp-outbound-channel-adapter">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. If the connection factory has a type 'client',
the factory is 'owned' by this adapter. If it has a type 'server', it is owned by an inbound channel
adapter and this adapter will attempt to correlate messages to the connection on which an original
inbound message was received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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:complexType>
</xsd:element>
<xsd:element name="tcp-inbound-channel-adapter">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
@@ -250,6 +113,111 @@ adapter.
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-outbound-channel-adapter">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. If the connection factory has a type 'client',
the factory is 'owned' by this adapter. If it has a type 'server', it is owned by an inbound channel
adapter and this adapter will attempt to correlate messages to the connection on which an original
inbound message was received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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:complexType>
</xsd:element>
<xsd:element name="tcp-inbound-gateway">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an inbound adapter. The connection factory must be of type 'server'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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="reply-timeout" type="xsd:string"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="tcp-outbound-gateway">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ip.tcp.connection.ConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
A connection factory is needed by an outbound adapter. The connection factory must be of 'client'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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:complexType>
</xsd:element>
<xsd:element name="tcp-connection-factory">
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID"/>
@@ -384,10 +352,10 @@ its configuration specifies the number of threads.
</xsd:complexType>
</xsd:element>
<xsd:complexType name="ipAdapterType">
<xsd:complexType name="udpAdapterType">
<xsd:annotation>
<xsd:documentation>
Common configuration for IP-based adapters.
Common configuration for UDP-based adapters.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
@@ -401,87 +369,18 @@ its configuration specifies the number of threads.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="protocol" use="required">
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="tcp" />
<xsd:enumeration value="udp" />
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="using-nio" type="xsd:string" />
<xsd:attribute name="using-direct-buffers" type="xsd:string" />
<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:attribute name="receive-buffer-size" type="xsd:string" />
<xsd:attribute name="close" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation>
If true, for an outbound gateway, the socket will be closed after a response
is received and sent to the
reply channel. If false, the socket will remain open and be used to send the
next message. If true, for an inbound gateway, the socket will be closed after
the response is sent. If false, the socket will remain open and be used to
receive the next message.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
<xsd:complexType name="common-attributes">
<xsd:attribute name="id" type="xsd:ID"/>
<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="message-format">
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="length-header" />
<xsd:enumeration value="stx-etx" />
<xsd:enumeration value="crlf" />
<xsd:enumeration value="serialized" />
<xsd:enumeration value="custom" />
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="so-keep-alive" type="xsd:string" />
<xsd:attribute name="local-address" type="xsd:string">
<xsd:annotation>
<xsd:documentation>