diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index b5e2e8b15b..cdb144142d 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -34,7 +34,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter protected final int port; - protected volatile int soTimeout = 60 * 1000; + protected volatile int soTimeout = 0; protected volatile int soReceiveBufferSize = -1; diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java new file mode 100644 index 0000000000..db29fdc241 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java @@ -0,0 +1,102 @@ +/* + * 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 + * + */ +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 byte[] assembledData; + + /** + * Assembles data in format {@link #FORMAT_LENGTH_HEADER}. + * @return True when a message is completely assembled. + * @throws IOException + */ + protected abstract boolean assembleDataLengthFormat() throws IOException; + + /** + * Assembles data in format {@link #FORMAT_STX_ETX}. + * @return True when a message is completely assembled. + * @throws IOException + */ + protected abstract boolean assembleDataStxEtxFormat() throws IOException; + + /** + * Assembles data in format {@link #FORMAT_CRLF}. + * @return True when a message is completely assembled. + * @throws IOException + */ + protected abstract boolean assembleDataCrLfFormat() 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 boolean assembleDataCustomFormat() throws IOException; + + public boolean assembleData() throws IOException { + switch (this.messageFormat) { + case FORMAT_LENGTH_HEADER: + return assembleDataLengthFormat(); + case FORMAT_STX_ETX: + return assembleDataStxEtxFormat(); + case FORMAT_CRLF: + return assembleDataCrLfFormat(); + case FORMAT_CUSTOM: + return assembleDataCustomFormat(); + default: + throw new UnsupportedOperationException( + "Unsupported message format: " + messageFormat); + } + } + + /** + * @param messageFormat the messageFormat to set, + */ + public void setMessageFormat(int messageFormat) { + this.messageFormat = messageFormat; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java new file mode 100644 index 0000000000..42c3a9f352 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java @@ -0,0 +1,150 @@ +/* + * 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 java.util.concurrent.ThreadFactory; + +import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +/** + * 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 + * + */ +public abstract class AbstractTcpReceivingChannelAdapter extends + AbstractInternetProtocolReceivingChannelAdapter { + + protected volatile ThreadPoolTaskScheduler threadPoolTaskScheduler; + + protected volatile int poolSize = -1; + + protected volatile SocketMessageMapper mapper = new SocketMessageMapper(); + + protected volatile boolean soKeepAlive; + + protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER; + + protected Class customSocketReader; + + protected boolean usingDirectBuffers; + + /** + * Constructs a receiving channel adapter that listens on the port. + * @param port The port to listen on. + */ + public AbstractTcpReceivingChannelAdapter(int port) { + super(port); + } + + /* (non-Javadoc) + * @see org.springframework.integration.endpoint.AbstractEndpoint#doStop() + */ + @Override + protected void doStop() { + // TODO Auto-generated method stub + + } + + /** + * Creates the ThreadPoolTaskScheduler, if necessary, and calls + * {@link #server()}. + */ + public void run() { + if (logger.isDebugEnabled()) { + logger.debug(this.getClass().getSimpleName() + " running..."); + } + if (this.active && this.threadPoolTaskScheduler == null) { + this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); + this.threadPoolTaskScheduler.setThreadFactory(new ThreadFactory() { + public Thread newThread(Runnable runner) { + Thread thread = new Thread(runner); + thread.setName("TCP-Incoming-Msg-Handler"); + thread.setDaemon(true); + return thread; + } + }); + if (this.poolSize > 0) { + this.threadPoolTaskScheduler.setPoolSize(this.poolSize); + } + this.threadPoolTaskScheduler.initialize(); + } + 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); + + } + + /** + * @param soKeepAlive the soKeepAlive to set + */ + public void setSoKeepAlive(boolean soKeepAlive) { + this.soKeepAlive = soKeepAlive; + } + + /** + * @param messageFormat the messageFormat to set + */ + public void setMessageFormat(int messageFormat) { + this.messageFormat = messageFormat; + } + + /** + * @param customSocketReader the customSocketReader to set + * @throws ClassNotFoundException + */ + @SuppressWarnings("unchecked") + public void setCustomSocketReader(String customSocketReaderClassName) throws ClassNotFoundException { + this.customSocketReader = (Class) Class.forName(customSocketReaderClassName); + } + + /** + * @param usingDirectBuffers the usingDirectBuffers to set + */ + public void setUsingDirectBuffers(boolean usingDirectBuffers) { + this.usingDirectBuffers = usingDirectBuffers; + } + + /** + * @param poolSize the poolSize to set + */ + public void setPoolSize(int poolSize) { + this.poolSize = poolSize; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java new file mode 100644 index 0000000000..c64b09b565 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/MessageFormats.java @@ -0,0 +1,47 @@ +/* + * 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 '<length><message>'. + */ + public static final int FORMAT_LENGTH_HEADER = 1; + /** + * Message has format 'STX<message>ETX'. + */ + public static final int FORMAT_STX_ETX = 2; + /** + * Message has format '<message>\r\n'. + */ + public static final int FORMAT_CRLF = 3; + /** + * Message has custom format. + */ + public static final int FORMAT_CUSTOM = 99; + + public static final int STX = 0x02; + + public static final int ETX = 0x03; + +} \ No newline at end of file diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java new file mode 100644 index 0000000000..3d6bcb1845 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java @@ -0,0 +1,170 @@ +/* + * 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.InputStream; +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.adapter.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 + * + */ +public class NetSocketReader extends AbstractSocketReader { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected Socket socket; + + protected int receiveBufferSize = 1024 * 60; + + /** + * 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 boolean assembleDataLengthFormat() throws IOException { + byte[] lengthPart = new byte[4]; + read(lengthPart); + int messageLength = ByteBuffer.wrap(lengthPart).getInt(); + if (logger.isDebugEnabled()) { + logger.debug("Message length is " + messageLength); + } + byte[] messagePart = new byte[messageLength]; + read(messagePart); + assembledData = messagePart; + return true; + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat() + */ + @Override + protected boolean assembleDataStxEtxFormat() throws IOException { + InputStream inputStream = socket.getInputStream(); + if (inputStream.read() != STX) + throw new MessageMappingException("Expected STX to begin message"); + byte[] buffer = new byte[receiveBufferSize]; + int n = 0; + int bite; + while ((bite = inputStream.read()) != ETX) { + buffer[n++] = (byte) bite; + } + assembledData = new byte[n]; + System.arraycopy(buffer, 0, assembledData, 0, n); + return true; + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat() + */ + @Override + protected boolean assembleDataCrLfFormat() throws IOException { + InputStream inputStream = socket.getInputStream(); + byte[] buffer = new byte[receiveBufferSize]; + int n = 0; + int bite; + while (true) { + bite = inputStream.read(); + if (n > 0 && bite == '\n' && buffer[n-1] == '\r') + break; + buffer[n++] = (byte) bite; + }; + assembledData = new byte[n-1]; + System.arraycopy(buffer, 0, assembledData, 0, n-1); + return true; + } + + /** + * 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 boolean assembleDataCustomFormat() throws IOException { + throw new UnsupportedOperationException("Need to subclass for this format"); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.SocketReader#getAssembledData() + */ + public byte[] getAssembledData() { + byte[] assembledData = this.assembledData; + this.assembledData = null; + return assembledData; + } + + /** + * Reads data from the socket and puts the data in buffer. Blocks until + * buffer is full or a socket timeout occurs. + * @param buffer + * @throws IOException + */ + protected void read(byte[] buffer) throws IOException { + int lengthRead = 0; + int needed = buffer.length; + while (lengthRead < needed) { + int len; + len = socket.getInputStream().read(buffer, lengthRead, + needed - lengthRead); + if (len < 0) { + throw new IOException("EOF"); + } + lengthRead += len; + if (logger.isDebugEnabled()) { + logger.debug("Read " + len + " bytes, buffer is now at " + + lengthRead + " of " + + needed); + } + } + + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.SocketReader#getAddress() + */ + public InetAddress getAddress() { + return this.socket.getInetAddress(); + } + + /** + * Sets the socket. + * @param socket + */ + public void setSocket(Socket socket) { + this.socket = socket; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java new file mode 100644 index 0000000000..bcd1141191 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java @@ -0,0 +1,319 @@ +/* + * 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.nio.ByteBuffer; +import java.nio.channels.SocketChannel; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.adapter.MessageMappingException; + +/** + * A non-blocking SocketReader that reads from a {@link java.nio.channels.SocketChannel}. + * + * @author Gary Russell + * + */ +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 int receiveBufferSize = 1024 * 60; + + 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#read(java.nio.ByteBuffer, int) + */ + public byte[] getAssembledData() { + byte[] assembledData = this.assembledData; + this.assembledData = null; + return assembledData; + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.SocketReader#assembleData() + */ + @Override + public boolean assembleDataLengthFormat() { + try { + if (lengthPart == null) { + lengthPart = allocate(4); + } + if (lengthPart.hasRemaining()) { + readChannel(lengthPart); + return false; + } + if (dataPart == null) { + lengthPart.flip(); + int messageLength = lengthPart.getInt(); + if (logger.isDebugEnabled()) { + logger.debug("Message length is " + messageLength); + } + dataPart = allocate(messageLength); + } + if (dataPart.hasRemaining()) { + readChannel(dataPart); + if (dataPart.hasRemaining()) { + return false; + } + } + assembledData = dataPart.array(); + lengthPart = dataPart = null; + return true; + } catch (Exception e) { + e.printStackTrace(); + // TODO + try { + channel.close(); + } catch (IOException e1) { + // TODO Auto-generated catch block + e1.printStackTrace(); + } + throw new MessageMappingException("Message assembly exception", e); + } + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataStxEtxFormat() + */ + @Override + protected boolean assembleDataStxEtxFormat() throws IOException { + if (readChannelNonDeterministic()) { + byte bite = rawBuffer.get(); + int count = 0; + if (!building) { + if (bite != STX) { + throw new MessageMappingException("Expected STX, received " + Integer.toHexString(bite)); + } + building = true; + count++; + if (!rawBuffer.hasRemaining()) { + if (logger.isDebugEnabled()) { + logger.debug("Incomplete message, consumed 1 byte"); + } + return false; + } + } else { + if (bite == ETX) { + finishAssembly(); + return true; + } + buildBuffer.put(bite); + count++; + } + while (true) { + if (!rawBuffer.hasRemaining()) { + if (logger.isDebugEnabled()) { + logger.debug("Incomplete message, consumed " + count + " bytes"); + } + return false; + } + bite = rawBuffer.get(); + if (bite == ETX) { + break; + } + buildBuffer.put(bite); + count++; + } + if (logger.isDebugEnabled()) { + logger.debug("Consumed " + count + " bytes"); + } + finishAssembly(); + return true; + } else { + if (logger.isDebugEnabled()) { + logger.debug("Incomplete message, consumed 0 bytes"); + } + } + return false; + } + + /** + * + */ + private void finishAssembly() { + assembledData = new byte[buildBuffer.position()]; + System.arraycopy(buildBuffer.array(), 0, assembledData, 0, assembledData.length); + building = false; + buildBuffer.clear(); + logger.debug("Message assembly complete"); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCrLfFormat() + */ + @Override + protected boolean assembleDataCrLfFormat() throws IOException { + if (readChannelNonDeterministic()) { + int count = 0; + while (true) { + if (!rawBuffer.hasRemaining()) { + if (logger.isDebugEnabled()) { + logger.debug("Incomplete message, consumed " + count + " bytes"); + } + return false; + } + byte bite = rawBuffer.get(); + if (bite == '\n' && buildBuffer.position() > 0) { + buildBuffer.position(buildBuffer.position() - 1); + if (buildBuffer.get() == '\r') { + buildBuffer.position(buildBuffer.position() - 1); + break; + } + } + buildBuffer.put(bite); + count++; + } + if (logger.isDebugEnabled()) { + logger.debug("Consumed " + count + " bytes"); + } + finishAssembly(); + return true; + } else { + if (logger.isDebugEnabled()) { + logger.debug("Incomplete message, consumed 0 bytes"); + } + } + return false; + } + + /** + * 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 boolean 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 void readChannel(ByteBuffer buffer) throws IOException { + try { + int len = channel.read(buffer); + if (logger.isDebugEnabled()) { + logger.debug("Read " + len + " bytes, buffer is now at " + + buffer.position() + " of " + + buffer.capacity()); + } + } catch (IOException e) { + throw e; + } + } + + /** + * Reads data into the rawBuffer for non-deterministic algorithms. + * @return true If data is available. + * @throws IOException + */ + protected boolean readChannelNonDeterministic() throws IOException { + if (rawBuffer == null) { + rawBuffer = allocate(receiveBufferSize); + buildBuffer = ByteBuffer.allocate(receiveBufferSize); + } else if (rawBuffer.hasRemaining()) { + if (logger.isDebugEnabled()) { + logger.debug("Raw buffer has " + rawBuffer.remaining() + " remaining"); + } + return true; + } + rawBuffer.clear(); + int len = channel.read(rawBuffer); + if (len == 0) { + return false; + } + rawBuffer.flip(); + if (logger.isDebugEnabled()) { + logger.debug("Read " + rawBuffer.limit() + " into raw buffer"); + } + return true; + } + + /** + * Allocates a ByteBuffer of the requested length using normal or + * direct buffers, depending on the usingDirectBuffers field. + * @param length + * @return + */ + 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.SocketReader#getAddress() + */ + public InetAddress getAddress() { + return this.channel.socket().getInetAddress(); + } + + /** + * @return the useDirectBuffers + */ + public boolean isUsingDirectBuffers() { + return usingDirectBuffers; + } + + /** + * @param useDirectBuffers the useDirectBuffers to set + */ + public void setUsingDirectBuffers(boolean usingDirectBuffers) { + this.usingDirectBuffers = usingDirectBuffers; + } + + /** + * @param channel + */ + public void setChannel(SocketChannel channel) { + this.channel = channel; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java new file mode 100644 index 0000000000..b45685a2b1 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java @@ -0,0 +1,70 @@ +/* + * 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.springframework.integration.core.Message; +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.message.InboundMessageMapper; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.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, + OutboundMessageMapper { + + /* (non-Javadoc) + * @see org.springframework.integration.message.InboundMessageMapper#toMessage(java.lang.Object) + */ + public Message toMessage(SocketReader socketReader) throws Exception { + return fromRaw(socketReader); + } + + + /** + * @param socketReader + * @return + * @throws IOException + */ + private Message fromRaw(SocketReader socketReader) throws IOException { + byte[] payload = socketReader.getAssembledData(); + Message message = null; + if (payload != null && payload.length > 0) { + message = MessageBuilder.withPayload(payload) + .setHeader(IpHeaders.HOSTNAME, socketReader.getAddress().getHostName()) + .setHeader(IpHeaders.IP_ADDRESS, socketReader.getAddress().getHostAddress()) + .build(); + } + return message; + } + + /* (non-Javadoc) + * @see org.springframework.integration.message.OutboundMessageMapper#fromMessage(org.springframework.integration.core.Message) + */ + public SocketWriter fromMessage(Message message) throws Exception { + // TODO Auto-generated method stub + return null; + } + + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java new file mode 100644 index 0000000000..5375cb7e71 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketReader.java @@ -0,0 +1,52 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import java.io.IOException; +import java.net.InetAddress; + +/** + * 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 + * + */ +public interface SocketReader { + + /** + * Reads the data the socket and assembles + * packets of data into a complete message, depending on the format of that + * data. + * @return true when the message is assembled. + * @throws IOException + */ + public boolean 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 byte[] getAssembledData(); + + /** + * Returns the InetAddress of the underlying socket. + * @return The InetAddress. + */ + public InetAddress getAddress(); +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java new file mode 100644 index 0000000000..b9789f3035 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java @@ -0,0 +1,27 @@ +/* + * 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; + + +/** + * A general interface for writing to sockets. + * + * @author Gary Russell + * + */ +public interface SocketWriter { + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java new file mode 100644 index 0000000000..ed178798c0 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java @@ -0,0 +1,120 @@ +/* + * 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.ServerSocket; +import java.net.Socket; + +import javax.net.ServerSocketFactory; + +import org.springframework.integration.adapter.MessageMappingException; +import org.springframework.integration.core.Message; + +/** + * Tcp Receiving Channel adapter that uses a {@link java.net.Socket}. Each + * connected socket uses a dedicated thread so the pool size must be set + * accordingly. + * + * @author Gary Russell + * + */ +public class TcpNetReceivingChannelAdapter extends + AbstractTcpReceivingChannelAdapter { + + /** + * Constructs a TcpNetReceivingChannelAdapter that listens on the port. + * @param port The port. + */ + 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 (true) { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + while (true) { + final Socket socket = server.accept(); + setSocketOptions(socket); + this.threadPoolTaskScheduler.execute(new Runnable() { + public void run() { + handleSocket(socket); + }}); + } + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + + } + + } + + /** + * Constructs a {@link NetSocketReader} and calls its {@link NetSocketReader#assembledData} + * method repeatedly; for each assembled message, calls {@link #sendMessage(Message)} with + * the mapped message. + * + * @param socket + */ + protected void handleSocket(Socket socket) { + NetSocketReader reader = null; + if (messageFormat == MessageFormats.FORMAT_CUSTOM) { + try { + reader = (NetSocketReader) customSocketReader.newInstance(); + reader.setSocket(socket); + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + throw new MessageMappingException("Failed to instantiate custom reader", e); + } + } + else { + reader = new NetSocketReader(socket); + } + reader.setMessageFormat(messageFormat); + while (true) { + try { + if (reader.assembleData()) { + Message message = mapper.toMessage(reader); + if (message != null) { + sendMessage(message); + } + } + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + try { + socket.close(); + return; + } catch (IOException e1) { + // TODO Auto-generated catch block + e1.printStackTrace(); + } + } + } + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java new file mode 100644 index 0000000000..f263d690e8 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -0,0 +1,188 @@ +/* + * 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.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.core.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. + * + * @author Gary Russell + * + */ +public class TcpNioReceivingChannelAdapter extends + AbstractTcpReceivingChannelAdapter { + + /** + * 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 { + final ServerSocketChannel server = ServerSocketChannel.open(); + server.configureBlocking(false); + server.socket().bind(new InetSocketAddress(port)); + final Selector selector = Selector.open(); + server.register(selector, SelectionKey.OP_ACCEPT); + doSelect(server, selector); + + } catch (IOException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + + /** + * 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 (true) { + int selectionCount = selector.select(); + if (logger.isDebugEnabled()) + logger.debug("SelectionCount: " + selectionCount); + if (selectionCount > 0) { + Set keys = selector.selectedKeys(); + Iterator 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); + key.attach(reader); + } + this.threadPoolTaskScheduler.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) { + NioSocketReader reader = null; + SocketChannel channel = (SocketChannel) key.channel(); + if (messageFormat == MessageFormats.FORMAT_CUSTOM) { + try { + reader = (NioSocketReader) customSocketReader.newInstance(); + reader.setChannel(channel); + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } else { + reader = new NioSocketReader(channel); + } + reader.setUsingDirectBuffers(usingDirectBuffers); + reader.setMessageFormat(messageFormat); + 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. + * + * @param channel + */ + private void doRead(SelectionKey key) { + NioSocketReader reader = (NioSocketReader) key.attachment(); + try { + if (reader.assembleData()) { + Message message; + message = mapper.toMessage(reader); + if (message != null) { + sendMessage(message); + } + } + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + try { + key.channel().close(); + } catch (IOException e1) { + // TODO Auto-generated catch block + e1.printStackTrace(); + } + } + } + +} + \ No newline at end of file diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketReader.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketReader.java new file mode 100644 index 0000000000..d2c160af20 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketReader.java @@ -0,0 +1,53 @@ +/* + * 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; + +/** + * Reads messages that are exactly 24 bytes long. + * + * @author Gary Russell + * + */ +public class CustomNetSocketReader extends NetSocketReader { + + public CustomNetSocketReader() { + super(null); + } + + /** + * @param socket + */ + public CustomNetSocketReader(Socket socket) { + super(socket); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat() + */ + @Override + protected boolean assembleDataCustomFormat() throws IOException { + byte[] buff = new byte[24]; + read(buff); + assembledData = buff; + return true; + } + + + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketReader.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketReader.java new file mode 100644 index 0000000000..36d3efc88a --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketReader.java @@ -0,0 +1,62 @@ +/* + * 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.nio.ByteBuffer; +import java.nio.channels.SocketChannel; + +/** + * Reads messages that are exactly 24 bytes long. + * + * @author Gary Russell + * + */ +public class CustomNioSocketReader extends NioSocketReader { + + private ByteBuffer buffer; + + public CustomNioSocketReader() { + super(null); + } + + /** + * @param socket + */ + public CustomNioSocketReader(SocketChannel channel) { + super(channel); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.NetSocketReader#assembleDataCustomFormat() + */ + @Override + protected boolean assembleDataCustomFormat() throws IOException { + if (buffer == null) { + buffer = allocate(24); + } + readChannel(buffer); + if (buffer.hasRemaining()) { + return false; + } + assembledData = buffer.array(); + buffer = null; + return true; + } + + + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java new file mode 100644 index 0000000000..f005fc42d5 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java @@ -0,0 +1,120 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.net.ServerSocket; +import java.net.Socket; + +import javax.net.ServerSocketFactory; + +import org.junit.Test; + +/** + * @author Gary Russell + * + */ +public class NetSocketReaderTests { + + /** + * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader#readFully()}, + * using <length><message>. + */ + @Test + public void testReadLength() throws Exception { + int port = 23556; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Utils.testSendLength(port, null); + Socket socket = server.accept(); + socket.setSoTimeout(5000); + NetSocketReader reader = new NetSocketReader(socket); + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + } + else { + fail("Failed to assemble first message"); + } + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + } + else { + fail("Failed to assemble second message"); + } + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader#readFully()}, + * using STX<message>ETX + */ + @Test + public void testReadStxEtx() throws Exception { + int port = 23557; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Utils.testSendStxEtx(port, null); + Socket socket = server.accept(); + socket.setSoTimeout(5000); + NetSocketReader reader = new NetSocketReader(socket); + reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + } + else { + fail("Failed to assemble first message"); + } + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + } + else { + fail("Failed to assemble second message"); + } + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader#readFully()}, + * using STX<message>ETX + */ + @Test + public void testReadCrLf() throws Exception { + int port = 23558; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Utils.testSendCrLf(port, null); + Socket socket = server.accept(); + socket.setSoTimeout(5000); + NetSocketReader reader = new NetSocketReader(socket); + reader.setMessageFormat(MessageFormats.FORMAT_CRLF); + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + } + else { + fail("Failed to assemble first message"); + } + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + } + else { + fail("Failed to assemble second message"); + } + } + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java new file mode 100644 index 0000000000..d7386263f9 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java @@ -0,0 +1,271 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.net.InetSocketAddress; +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 java.util.concurrent.CountDownLatch; + +import org.junit.Test; + +/** + * @author Gary Russell + * + */ +public class NioSocketReaderTests { + + private CountDownLatch latch = new CountDownLatch(1); + /** + * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}. + */ + @Test + public void testReadLength() throws Exception { + ServerSocketChannel server = ServerSocketChannel.open(); + server.configureBlocking(false); + int port = 23456; + server.socket().bind(new InetSocketAddress(port)); + final Selector selector = Selector.open(); + server.register(selector, SelectionKey.OP_ACCEPT); + + // Fire up the sender. + Utils.testSendLength(port, latch); + + if(selector.select(10000) <= 0) { + fail("Socket failed to connect"); + } + Set keys = selector.selectedKeys(); + Iterator iterator = keys.iterator(); + SocketChannel channel = null; + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isAcceptable()) { + channel = server.accept(); + channel.configureBlocking(false); + channel.register(selector, SelectionKey.OP_READ); + } + else { + fail("Unexpected key: " + key); + } + } + NioSocketReader reader = new NioSocketReader(channel); + int count = 0; + while(selector.select(1000) > 0) { + keys = selector.selectedKeys(); + iterator = keys.iterator(); + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isReadable()) { + assertEquals(channel, key.channel()); + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + count++; + } + latch.countDown(); + } + else { + fail("Unexpected key: " + key); + } + } + } + assertEquals("Did not receive data", 2, count); + } + + @Test + public void testFragmented() throws Exception { + ServerSocketChannel server = ServerSocketChannel.open(); + server.configureBlocking(false); + int port = 23457; + server.socket().bind(new InetSocketAddress(port)); + final Selector selector = Selector.open(); + server.register(selector, SelectionKey.OP_ACCEPT); + + // Fire up the sender. + Utils.testSendFragmented(port); + + if(selector.select(10000) <= 0) { + fail("Socket failed to connect"); + } + Set keys = selector.selectedKeys(); + Iterator iterator = keys.iterator(); + SocketChannel channel = null; + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isAcceptable()) { + channel = server.accept(); + channel.configureBlocking(false); + channel.register(selector, SelectionKey.OP_READ); + } + else { + fail("Unexpected key: " + key); + } + } + NioSocketReader reader = new NioSocketReader(channel); + boolean done = false; + while(selector.select(1000) > 0) { + keys = selector.selectedKeys(); + iterator = keys.iterator(); + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isReadable()) { + assertEquals(channel, key.channel()); + if (reader.assembleData()) { + assertEquals("Data", "xx", + new String(reader.getAssembledData())); + done = true; + } + latch.countDown(); + } + else { + fail("Unexpected key: " + key); + } + } + } + assertTrue("Did not receive data", done); + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}. + */ + @Test + public void testReadStxEtx() throws Exception { + ServerSocketChannel server = ServerSocketChannel.open(); + server.configureBlocking(false); + int port = 23458; + server.socket().bind(new InetSocketAddress(port)); + final Selector selector = Selector.open(); + server.register(selector, SelectionKey.OP_ACCEPT); + + // Fire up the sender. + Utils.testSendStxEtx(port, latch); + + if(selector.select(10000) <= 0) { + fail("Socket failed to connect"); + } + Set keys = selector.selectedKeys(); + Iterator iterator = keys.iterator(); + SocketChannel channel = null; + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isAcceptable()) { + channel = server.accept(); + channel.configureBlocking(false); + channel.register(selector, SelectionKey.OP_READ); + } + else { + fail("Unexpected key: " + key); + } + } + NioSocketReader reader = new NioSocketReader(channel); + reader.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + int count = 0; + while(selector.select(1000) > 0) { + keys = selector.selectedKeys(); + iterator = keys.iterator(); + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isReadable()) { + assertEquals(channel, key.channel()); + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + count++; + } + latch.countDown(); + } + else { + fail("Unexpected key: " + key); + } + } + } + assertEquals("Did not receive data", 2, count); + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}. + */ + @Test + public void testReadCrLf() throws Exception { + ServerSocketChannel server = ServerSocketChannel.open(); + server.configureBlocking(false); + int port = 23459; + server.socket().bind(new InetSocketAddress(port)); + final Selector selector = Selector.open(); + server.register(selector, SelectionKey.OP_ACCEPT); + + // Fire up the sender. + Utils.testSendCrLf(port, latch); + + if(selector.select(10000) <= 0) { + fail("Socket failed to connect"); + } + Set keys = selector.selectedKeys(); + Iterator iterator = keys.iterator(); + SocketChannel channel = null; + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isAcceptable()) { + channel = server.accept(); + channel.configureBlocking(false); + channel.register(selector, SelectionKey.OP_READ); + } + else { + fail("Unexpected key: " + key); + } + } + NioSocketReader reader = new NioSocketReader(channel); + reader.setMessageFormat(MessageFormats.FORMAT_CRLF); + int count = 0; + while(selector.select(1000) > 0) { + keys = selector.selectedKeys(); + iterator = keys.iterator(); + while (iterator.hasNext()) { + SelectionKey key = iterator.next(); + iterator.remove(); + if (key.isReadable()) { + assertEquals(channel, key.channel()); + if (reader.assembleData()) { + assertEquals("Data", Utils.TEST_STRING + Utils.TEST_STRING, + new String(reader.getAssembledData())); + count++; + } + latch.countDown(); + } + else { + fail("Unexpected key: " + key); + } + } + } + assertEquals("Did not receive data", 2, count); + } + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java new file mode 100644 index 0000000000..9844a151c2 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java @@ -0,0 +1,98 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.net.InetAddress; +import java.net.UnknownHostException; + +import org.junit.Ignore; +import org.junit.Test; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.IpHeaders; + +/** + * @author Gary Russell + * + */ +public class SocketMessageMapperTests { + + /** + * + */ + private static final String TEST_PAYLOAD = "abcdefghijkl"; + + /** + * Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#toMessage(org.springframework.integration.ip.tcp.SocketReader)}. + * Tests segmented reads into the payload and verifies reassembly. + */ + @Test + public void testToMessage() throws Exception { + SocketMessageMapper mapper = new SocketMessageMapper(); + Message message = mapper.toMessage(new StubSocketReader()); + assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); + assertEquals(InetAddress.getLocalHost().getHostName(), message + .getHeaders().get(IpHeaders.HOSTNAME)); + assertEquals(InetAddress.getLocalHost().getHostAddress(), message + .getHeaders().get(IpHeaders.IP_ADDRESS)); + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#fromMessage(org.springframework.integration.core.Message)}. + */ + @Test + @Ignore + public void testFromMessage() { + fail("Not yet implemented"); + } + + + private class StubSocketReader implements SocketReader { + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.SocketReader#getAddress() + */ + public InetAddress getAddress() { + try { + return InetAddress.getLocalHost(); + } catch (UnknownHostException e) { + fail("Unexpected Exception: " + e.getMessage()); + } + return null; + } + + public byte[] getAssembledData() { + return TEST_PAYLOAD.getBytes(); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.SocketReader#assembleData() + */ + public boolean assembleData() { + return false; + } + + + } + + private class StubSocketWriter implements SocketWriter { + + + } + +} \ No newline at end of file diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java new file mode 100644 index 0000000000..9833b62263 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -0,0 +1,142 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +/** + * @author Gary Russell + * + */ +public class TcpReceivingChannelAdapterTests { + + /** + * Test method for {@link org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter#run()}. + */ + @Test + public void testNet() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = 12345; + AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + Thread.sleep(2000); // wait for server to start listening + Utils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice + Thread.sleep(2000); // wait for asynch processing + Message message = channel.receive(0); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + new String((byte[])message.getPayload())); + message = channel.receive(0); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + new String((byte[])message.getPayload())); + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter#run()}. + * Verifies operation of custom message formats. + */ + @Test + public void testNetCustom() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = 12346; + AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + adapter.setCustomSocketReader("org.springframework.integration.ip.tcp.CustomNetSocketReader"); + adapter.setMessageFormat(MessageFormats.FORMAT_CUSTOM); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + Thread.sleep(2000); // wait for server to start listening + Utils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice + Thread.sleep(4000); // wait for asynch processing + Message message = channel.receive(0); + assertNotNull(message); + assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + new String((byte[])message.getPayload())); + message = channel.receive(0); + assertNotNull(message); + assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + new String((byte[])message.getPayload())); + } + + + /** + * Test method for {@link org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter#run()}. + */ + @Test + public void testNio() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = 12355; + TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + Thread.sleep(2000); // wait for server to start listening + Utils.testSendLength(port, null); //sends 2 copies of TEST_STRING twice + Thread.sleep(2000); // wait for asynch processing + Message message = channel.receive(0); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + new String((byte[])message.getPayload())); + message = channel.receive(0); + assertNotNull(message); + assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, + new String((byte[])message.getPayload())); + } + + /** + * Test method for {@link org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter#run()}. + * Verifies operation of custom message formats. */ + @Test + public void testNioCustom() throws Exception { + QueueChannel channel = new QueueChannel(2); + int port = 12356; + TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); + adapter.setOutputChannel(channel); + adapter.setCustomSocketReader("org.springframework.integration.ip.tcp.CustomNioSocketReader"); + adapter.setMessageFormat(MessageFormats.FORMAT_CUSTOM); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + adapter.setTaskScheduler(taskScheduler); + adapter.start(); + Thread.sleep(2000); // wait for server to start listening + Utils.testSendStxEtx(port, null); //sends 2 copies of TEST_STRING twice + Thread.sleep(4000); // wait for asynch processing + Message message = channel.receive(0); + assertNotNull(message); + assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + new String((byte[])message.getPayload())); + message = channel.receive(0); + assertNotNull(message); + assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", + new String((byte[])message.getPayload())); + } + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java new file mode 100644 index 0000000000..7075a5fbba --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java @@ -0,0 +1,167 @@ +/* + * 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.OutputStream; +import java.net.InetAddress; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.util.concurrent.CountDownLatch; + +/** + * TCP/IP Test utilities. + * + * @author Gary Russell + * + */ +public class Utils { + + public static final String TEST_STRING = "TestMessage"; + + /** + * Sends a message in two chunks with a preceding length. Two such messages are sent. + * @param latch If not null, await until counted down before sending second chunk. + */ + public static void testSendLength(final int port, final CountDownLatch latch) { + Thread thread = new Thread(new Runnable() { + public void run() { + try { + Socket socket = new Socket(InetAddress.getByName("localhost"), port); + for (int i = 0; i < 2; i++) { + byte[] len = new byte[4]; + ByteBuffer.wrap(len).putInt(TEST_STRING.length() * 2); + socket.getOutputStream().write(len); + socket.getOutputStream().write(TEST_STRING.getBytes()); + System.out.println(i + " Wrote first part"); + if (latch != null) { + latch.await(); + } + Thread.sleep(500); + // send the second chunk + socket.getOutputStream().write(TEST_STRING.getBytes()); + System.out.println(i + " Wrote second part"); + } + Thread.sleep(1000000000L); // wait forever, but we're a daemon + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } + + /** + * Test for reassembly of completely fragmented message; sends + * 6 bytes 500ms apart. + * @param os + * @param b + * @throws Exception + */ + public static void testSendFragmented(final int port) { + Thread thread = new Thread(new Runnable() { + public void run() { + try { + Socket socket = new Socket(InetAddress.getByName("localhost"), port); + OutputStream os = socket.getOutputStream(); + writeByte(os, 0); + writeByte(os, 0); + writeByte(os, 0); + writeByte(os, 2); + writeByte(os, 'x'); + writeByte(os, 'x'); + Thread.sleep(1000000000L); // wait forever, but we're a daemon + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } + + private static void writeByte(OutputStream os, int b) throws Exception { + os.write(b); + System.out.printf("Wrote 0x%x\n", b); + Thread.sleep(500); + } + + /** + * Sends a STX/ETX message in two chunks. Two such messages are sent. + * @param latch If not null, await until counted down before sending second chunk. + */ + public static void testSendStxEtx(final int port, final CountDownLatch latch) { + Thread thread = new Thread(new Runnable() { + public void run() { + try { + Socket socket = new Socket(InetAddress.getByName("localhost"), port); + OutputStream outputStream = socket.getOutputStream(); + for (int i = 0; i < 2; i++) { + writeByte(outputStream, 0x02); + outputStream.write(TEST_STRING.getBytes()); + System.out.println(i + " Wrote first part"); + if (latch != null) { + latch.await(); + } + Thread.sleep(500); + // send the second chunk + outputStream.write(TEST_STRING.getBytes()); + System.out.println(i + " Wrote second part"); + writeByte(outputStream, 0x03); + } + Thread.sleep(1000000000L); // wait forever, but we're a daemon + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } + + /** + * Sends a message +CRLF in two chunks. Two such messages are sent. + * @param latch If not null, await until counted down before sending second chunk. + */ + public static void testSendCrLf(final int port, final CountDownLatch latch) { + Thread thread = new Thread(new Runnable() { + public void run() { + try { + Socket socket = new Socket(InetAddress.getByName("localhost"), port); + OutputStream outputStream = socket.getOutputStream(); + for (int i = 0; i < 2; i++) { + outputStream.write(TEST_STRING.getBytes()); + System.out.println(i + " Wrote first part"); + if (latch != null) { + latch.await(); + } + Thread.sleep(500); + // send the second chunk + outputStream.write(TEST_STRING.getBytes()); + System.out.println(i + " Wrote second part"); + writeByte(outputStream, '\r'); + writeByte(outputStream, '\n'); + } + Thread.sleep(1000000000L); // wait forever, but we're a daemon + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + thread.setDaemon(true); + thread.start(); + } +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/DatagramPacketMessageMapperTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java similarity index 98% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/DatagramPacketMessageMapperTests.java rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java index 9106fc7b9d..401b9a0b34 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/DatagramPacketMessageMapperTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.ip; +package org.springframework.integration.ip.udp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/DatagramPacketSendingHandlerTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java similarity index 99% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/DatagramPacketSendingHandlerTests.java rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java index bfd7bd5b03..7838c1d577 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/DatagramPacketSendingHandlerTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.ip; +package org.springframework.integration.ip.udp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/UdpMulticastEndToEndTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java similarity index 97% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/UdpMulticastEndToEndTests.java rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java index c1ff6868dd..36788beca8 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/UdpMulticastEndToEndTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpMulticastEndToEndTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.ip; +package org.springframework.integration.ip.udp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -32,6 +32,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.ip.StdOutCatcher; import org.springframework.integration.message.StringMessage; /** diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/UdpUnicastEndToEndTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java similarity index 96% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/UdpUnicastEndToEndTests.java rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java index 154eaf7f9a..e49dfd2aaf 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/UdpUnicastEndToEndTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/UdpUnicastEndToEndTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.ip; +package org.springframework.integration.ip.udp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -30,15 +30,15 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.ip.StdOutCatcher; import org.springframework.integration.message.StringMessage; /** * Sends and receives a simple message through to the Udp channel adapters. * If run as a JUnit just sends one message and terminates (see console). - * TODO: Use a custom output stream and catch output to verify. * * If run from main(), - * hangs around for a couple of minutes to allow console interaction (enter a message on the + * hangs around for a couple of minutes to allow console interaction - enter a message on the * console and you should see it go through the outbound context, over UDP, and * received in the other context (and written back to the console). * diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-common-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml similarity index 95% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-common-context.xml rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml index 328e8a2bd7..ea1a5ab78e 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-common-context.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-common-context.xml @@ -23,7 +23,7 @@ - + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-in-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml similarity index 100% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-in-context.xml rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-context.xml diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-in-multicast-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-multicast-context.xml similarity index 100% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-in-multicast-context.xml rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-in-multicast-context.xml diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-out-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml similarity index 100% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-out-context.xml rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-out-multicast-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml similarity index 100% rename from org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/testIp-out-multicast-context.xml rename to org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-multicast-context.xml diff --git a/org.springframework.integration.ip/template.mf b/org.springframework.integration.ip/template.mf index 57fb593681..a0ae1b519b 100644 --- a/org.springframework.integration.ip/template.mf +++ b/org.springframework.integration.ip/template.mf @@ -10,4 +10,5 @@ Import-Template: org.springframework.core.*;version="[3.0.0, 4.0.0)", org.springframework.scheduling.*;version="[3.0.0, 4.0.0)", org.springframework.util;version="[3.0.0, 4.0.0)", - org.w3c.dom.*;version="0" + org.w3c.dom.*;version="0", + javax.net.*;version="0"