From 6e2295b6253104a67489601f678dbbf08b0b568c Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 25 Feb 2010 00:59:13 +0000 Subject: [PATCH] INT-781 Inbound TCP Adapters & Tests - no namespace support yet. --- ...InternetProtocolSendingMessageHandler.java | 9 + .../ip/tcp/AbstractSocketReader.java | 34 +- .../ip/tcp/AbstractSocketWriter.java | 110 +++++++ .../AbstractTcpReceivingChannelAdapter.java | 22 +- .../tcp/AbstractTcpSendingMessageHandler.java | 190 +++++++++++ .../integration/ip/tcp/NetSocketReader.java | 21 +- .../integration/ip/tcp/NetSocketWriter.java | 91 ++++++ .../integration/ip/tcp/NioSocketReader.java | 87 +++-- .../integration/ip/tcp/NioSocketWriter.java | 174 ++++++++++ .../ip/tcp/SocketMessageMapper.java | 47 ++- .../integration/ip/tcp/SocketWriter.java | 11 + .../ip/tcp/TcpNetReceivingChannelAdapter.java | 30 +- .../ip/tcp/TcpNetSendingMessageHandler.java | 85 +++++ .../ip/tcp/TcpNioReceivingChannelAdapter.java | 42 ++- .../ip/tcp/TcpNioSendingMessageHandler.java | 90 ++++++ .../udp/MulticastSendingMessageHandler.java | 2 +- .../ip/udp/UnicastSendingMessageHandler.java | 5 +- .../ip/tcp/CustomNetSocketWriter.java | 53 ++++ .../ip/tcp/CustomNioSocketWriter.java | 57 ++++ .../ip/tcp/NetSocketReaderTests.java | 9 +- .../ip/tcp/NetSocketWriterTests.java | 143 +++++++++ .../ip/tcp/NioSocketReaderTests.java | 12 +- .../ip/tcp/NioSocketWriterTests.java | 244 ++++++++++++++ .../ip/tcp/SocketMessageMapperTests.java | 18 +- .../tcp/TcpReceivingChannelAdapterTests.java | 17 +- .../ip/tcp/TcpSendingMessageHandlerTests.java | 298 ++++++++++++++++++ .../integration/ip/tcp/Utils.java | 14 + 27 files changed, 1772 insertions(+), 143 deletions(-) create mode 100644 org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java create mode 100644 org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java create mode 100644 org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java create mode 100644 org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java create mode 100644 org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java create mode 100644 org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java create mode 100644 org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java create mode 100644 org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java create mode 100644 org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java create mode 100644 org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java create mode 100644 org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java index 940449ac37..286f6f36ea 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip; import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.util.concurrent.ExecutorService; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -35,6 +36,10 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M protected final Log logger = LogFactory.getLog(getClass()); protected final SocketAddress destinationAddress; + + protected final String host; + + protected final int port; protected int soReceiveBufferSize = -1; @@ -42,10 +47,14 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M protected volatile int soTimeout = -1; + protected volatile ExecutorService executorService; + public AbstractInternetProtocolSendingMessageHandler(String host, int port) { Assert.notNull(host, "host must not be null"); this.destinationAddress = new InetSocketAddress(host, port); + this.host = host; + this.port = port; } 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 index db29fdc241..dcd5959b56 100644 --- 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 @@ -77,21 +77,31 @@ public abstract class AbstractSocketReader implements SocketReader, MessageForma 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); + try { + 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); + } + } catch (IOException e) { + doClose(); + throw e; } } + /** + * Called after an exception; close the transport. + */ + protected abstract void doClose(); + /** * @param messageFormat the messageFormat to set, */ diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java new file mode 100644 index 0000000000..417ce080fe --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java @@ -0,0 +1,110 @@ +/* + * 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 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(byte[])} which is invoked by {@link #write(byte[])} + * when the format is {@link MessageFormats#FORMAT_CUSTOM}. + * @author Gary Russell + * + */ +public abstract class AbstractSocketWriter implements SocketWriter, MessageFormats { + + protected int messageFormat = FORMAT_LENGTH_HEADER; + + /* + * @see org.springframework.integration.ip.tcp.SocketWriter#write(byte[]) + */ + public synchronized void write(byte[] bytes) throws IOException { + try { + switch (this.messageFormat) { + case FORMAT_LENGTH_HEADER: + writeLengthFormat(bytes); + return; + case FORMAT_STX_ETX: + writeStxEtxFormat(bytes); + return; + case FORMAT_CRLF: + writeCrLfFormat(bytes); + return; + case FORMAT_CUSTOM: + writeCustomFormat(bytes); + 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'). + * @param bytes + */ + protected abstract void writeCrLfFormat(byte[] bytes) throws IOException; + + /** + * Write the data using some custom protocol. + * @param bytes + */ + protected abstract void writeCustomFormat(byte[] bytes) throws IOException; + + /** + * @param messageFormat the messageFormat to set + */ + public void setMessageFormat(int messageFormat) { + this.messageFormat = messageFormat; + } + +} \ No newline at end of file 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 index ae07a830f5..b705835fb0 100644 --- 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 @@ -43,10 +43,6 @@ public abstract class AbstractTcpReceivingChannelAdapter extends 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. @@ -102,6 +98,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends } /** + * @see {@link Socket#setKeepAlive(boolean)}. * @param soKeepAlive the soKeepAlive to set */ public void setSoKeepAlive(boolean soKeepAlive) { @@ -109,28 +106,13 @@ public abstract class AbstractTcpReceivingChannelAdapter extends } /** + * @See {@link MessageFormats} * @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 setCustomSocketReaderClassName(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 */ diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java new file mode 100644 index 0000000000..33529a9e1a --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java @@ -0,0 +1,190 @@ +/* + * 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.Executors; +import java.util.concurrent.ThreadFactory; + +import org.springframework.integration.adapter.MessageMappingException; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.MessageRejectedException; + +/** + * 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; + + protected boolean blockingWrite = false; + + /** + * 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. If blockingWrite is true, the write to the socket + * will occur on the caller's thread. Otherwise, the method will return + * immediately and the write will occur on a separate thread. + * + * @see org.springframework.integration.message.MessageHandler#handleMessage(org.springframework.integration.core.Message) + */ + public void handleMessage(final Message message) throws MessageRejectedException, + MessageHandlingException, MessageDeliveryException { + if (blockingWrite) { + doWrite(message); + return; + } + if (this.executorService == null) { + this.executorService = Executors + .newSingleThreadExecutor(new ThreadFactory() { + public Thread newThread(Runnable runner) { + Thread thread = new Thread(runner); + thread.setName("UDP-Ack-Handler"); + thread.setDaemon(true); + return thread; + } + }); + } + executorService.execute(new Runnable() { + public void run() { + doWrite(message); + } + }); + } + + /** + * Method that actually does the write. + * @param message The message to write. + */ + protected void doWrite(Message message) { + try { + byte[] bytes = mapper.fromMessage(message); + this.getWriter().write(bytes); + } catch (Exception e) { + writer = null; + throw new MessageMappingException("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; + } + + /** + * If true, socket writes will occur on the caller's thread. + * @param blockingWrite the blockingWrite to set + */ + public void setBlockingWrite(boolean blockingWrite) { + this.blockingWrite = blockingWrite; + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java index 3d6bcb1845..17c9f707aa 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java @@ -107,7 +107,7 @@ public class NetSocketReader extends AbstractSocketReader { /** * Throws {@link UnsupportedOperationException}; custom implementations can - * subclass this class and provide an implementation. + * subclass this class and provide an implementation for this method. * @throws IOException * @see org.springframework.integration.ip.tcp.AbstractSocketReader#assembleDataCustomFormat(). * @@ -152,6 +152,17 @@ public class NetSocketReader extends AbstractSocketReader { } + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketReader#doClose() + */ + @Override + protected void doClose() { + try { + socket.close(); + } catch (IOException e) {} + } + + /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.SocketReader#getAddress() */ @@ -159,12 +170,4 @@ public class NetSocketReader extends AbstractSocketReader { 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/NetSocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java new file mode 100644 index 0000000000..d24603b554 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java @@ -0,0 +1,91 @@ +/* + * 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.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 + * + */ +public class NetSocketWriter extends AbstractSocketWriter { + + protected Socket socket; + /** + * 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 = socket.getOutputStream(); + outputStream.write(bytes); + outputStream.write('\r'); + outputStream.write('\n'); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[]) + */ + @Override + protected void writeCustomFormat(byte[] bytes) 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 = socket.getOutputStream(); + outputStream.write(lengthPart.array()); + outputStream.write(bytes); + } + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeStxEtxFormat(byte[]) + */ + @Override + protected void writeStxEtxFormat(byte[] bytes) throws IOException { + OutputStream outputStream = socket.getOutputStream(); + outputStream.write(STX); + outputStream.write(bytes); + outputStream.write(ETX); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#doClose() + */ + @Override + protected void doClose() { + try { + socket.close(); + } catch (IOException e) {} + } +} 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 index ea56b58ab5..f705aa54cb 100644 --- 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 @@ -71,50 +71,38 @@ public class NioSocketReader extends AbstractSocketReader { * @see org.springframework.integration.ip.tcp.SocketReader#assembleData() */ @Override - public boolean assembleDataLengthFormat() { - try { - if (lengthPart == null) { - lengthPart = allocate(4); + public boolean assembleDataLengthFormat() throws IOException { + 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); } - if (lengthPart.hasRemaining()) { - readChannel(lengthPart); + dataPart = allocate(messageLength); + } + if (dataPart.hasRemaining()) { + readChannel(dataPart); + if (dataPart.hasRemaining()) { 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; - } - } - if (usingDirectBuffers) { - byte[] assembledData = new byte[dataPart.capacity()]; - dataPart.flip(); - dataPart.get(assembledData); - this.assembledData = assembledData; - } else { - 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); + } + if (usingDirectBuffers) { + byte[] assembledData = new byte[dataPart.capacity()]; + dataPart.flip(); + dataPart.get(assembledData); + this.assembledData = assembledData; + } else { + assembledData = dataPart.array(); } + lengthPart = dataPart = null; + return true; } /* (non-Javadoc) @@ -294,6 +282,18 @@ public class NioSocketReader extends AbstractSocketReader { } return buffer; } + + + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketReader#doClose() + */ + @Override + protected void doClose() { + try { + channel.close(); + } catch (IOException e) {} + } /* (non-Javadoc) * @see org.springframework.integration.ip.tcp.SocketReader#getAddress() @@ -316,11 +316,4 @@ public class NioSocketReader extends AbstractSocketReader { 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/NioSocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java new file mode 100644 index 0000000000..f520665a12 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java @@ -0,0 +1,174 @@ +/* + * 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; + +/** + * A {@link SocketWriter} that writes to a {@link java.nio.channels.SocketChannel}. The + * data is wrapped in a wire protocol based on the messageFormat property. + * + * @author Gary Russell + * + */ +public class NioSocketWriter extends AbstractSocketWriter { + + protected SocketChannel channel; + + /** + * If true, direct buffers are used. + * @see {@link 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; + + /** + * @param socket + */ + public NioSocketWriter(SocketChannel channel) { + this.channel = channel; + } + + /** + * @param usingDirectBuffers the usingDirectBuffers to set + */ + public void setUsingDirectBuffers(boolean usingDirectBuffers) { + this.usingDirectBuffers = usingDirectBuffers; + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCrLfFormat(byte[]) + */ + @Override + protected void writeCrLfFormat(byte[] bytes) throws IOException { + ByteBuffer buffer = null; + if (usingDirectBuffers) { + buffer = ByteBuffer.allocateDirect(bytes.length + 2); + buffer.put(bytes); + buffer.put((byte) '\r'); + buffer.put((byte) '\n'); + buffer.flip(); + channel.write(buffer); + return; + } + if (crLfPart == null) { + crLfPart = ByteBuffer.allocate(2); + crLfPart.put((byte) '\r'); + crLfPart.put((byte) '\n'); + } + channel.write(ByteBuffer.wrap(bytes)); + crLfPart.flip(); + channel.write(crLfPart); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCustomFormat(byte[]) + */ + @Override + protected void writeCustomFormat(byte[] bytes) throws IOException { + // TODO Auto-generated method stub + + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeLengthFormat(byte[]) + */ + @Override + protected void writeLengthFormat(byte[] bytes) throws IOException { + ByteBuffer buffer = null; + if (usingDirectBuffers) { + buffer = ByteBuffer.allocateDirect(bytes.length + 4); + buffer.putInt(bytes.length); + buffer.put(bytes); + buffer.flip(); + channel.write(buffer); + return; + } + if (lengthPart == null) { + lengthPart = ByteBuffer.allocate(4); + } + lengthPart.putInt(bytes.length); + lengthPart.flip(); + channel.write(lengthPart); + 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 (usingDirectBuffers) { + buffer = ByteBuffer.allocateDirect(bytes.length + 2); + buffer.put((byte) STX); + buffer.put(bytes); + buffer.put((byte) ETX); + buffer.flip(); + channel.write(buffer); + return; + } + if (stxPart == null) { + stxPart = ByteBuffer.allocate(1); + stxPart.put((byte) STX); + etxPart = ByteBuffer.allocate(1); + etxPart.put((byte) ETX); + } + stxPart.flip(); + channel.write(stxPart); + channel.write(ByteBuffer.wrap(bytes)); + etxPart.flip(); + channel.write(etxPart); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.AbstractSocketWriter#doClose() + */ + @Override + protected void doClose() { + try { + channel.close(); + } catch (IOException e) {} + } + + +} 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 index b45685a2b1..f1964c99a6 100644 --- 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 @@ -16,11 +16,13 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.io.UnsupportedEncodingException; 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.MessageHandlingException; import org.springframework.integration.message.OutboundMessageMapper; /** @@ -31,8 +33,10 @@ import org.springframework.integration.message.OutboundMessageMapper; */ public class SocketMessageMapper implements InboundMessageMapper, - OutboundMessageMapper { + OutboundMessageMapper { + private volatile String charset = "UTF-8"; + /* (non-Javadoc) * @see org.springframework.integration.message.InboundMessageMapper#toMessage(java.lang.Object) */ @@ -42,6 +46,9 @@ public class SocketMessageMapper implements /** + * 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 @@ -61,10 +68,42 @@ public class SocketMessageMapper implements /* (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; + public byte[] fromMessage(Message message) throws Exception { + return getPayloadAsBytes(message); + } + + /** + * 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; + } + } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java index b9789f3035..2ae0003909 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java @@ -15,6 +15,8 @@ */ package org.springframework.integration.ip.tcp; +import java.io.IOException; + /** * A general interface for writing to sockets. @@ -24,4 +26,13 @@ package org.springframework.integration.ip.tcp; */ 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 bytes The bytes to write. + * @throws IOException + */ + void write(byte[] bytes) throws IOException; + } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java index 416cdb29b8..ae2ddca46e 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java @@ -16,11 +16,13 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.lang.reflect.Constructor; import java.net.ServerSocket; import java.net.Socket; import javax.net.ServerSocketFactory; +import org.springframework.beans.BeanUtils; import org.springframework.integration.adapter.MessageMappingException; import org.springframework.integration.core.Message; @@ -36,6 +38,7 @@ public class TcpNetReceivingChannelAdapter extends AbstractTcpReceivingChannelAdapter { protected ServerSocket serverSocket; + protected Class customSocketReader; /** * Constructs a TcpNetReceivingChannelAdapter that listens on the port. * @param port The port. @@ -93,8 +96,9 @@ public class TcpNetReceivingChannelAdapter extends NetSocketReader reader = null; if (messageFormat == MessageFormats.FORMAT_CUSTOM) { try { - reader = (NetSocketReader) customSocketReader.newInstance(); - reader.setSocket(socket); + Constructor ctor = + customSocketReader.getConstructor(Socket.class); + reader = BeanUtils.instantiateClass(ctor, socket); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); @@ -114,18 +118,11 @@ public class TcpNetReceivingChannelAdapter extends } } } catch (Exception e) { - // TODO Auto-generated catch block - e.printStackTrace(); - try { - socket.close(); - return; - } catch (IOException e1) { - // TODO Auto-generated catch block - e1.printStackTrace(); - } + return; } } } + @Override protected void doStop() { super.doStop(); @@ -137,4 +134,15 @@ public class TcpNetReceivingChannelAdapter extends } } + /** + * @param customSocketReader the customSocketReader to set + * @throws ClassNotFoundException + */ + @SuppressWarnings("unchecked") + public void setCustomSocketReaderClassName( + String customSocketReaderClassName) throws ClassNotFoundException { + this.customSocketReader = (Class) Class + .forName(customSocketReaderClassName); + } + } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java new file mode 100644 index 0000000000..83a5868e61 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java @@ -0,0 +1,85 @@ +/* + * 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.lang.reflect.Constructor; +import java.net.Socket; + +import javax.net.SocketFactory; + +import org.springframework.beans.BeanUtils; + + +/** + * TCP Sending Channel Adapter that that uses a {@link java.net.Socket}. + * @author Gary Russell + * + */ +public class TcpNetSendingMessageHandler extends + AbstractTcpSendingMessageHandler { + + protected Class customSocketWriter; + + /** + * 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; + + /** + * if + * @return the writer + */ + protected synchronized SocketWriter getWriter() { + if (writer == null) { + try { + this.socket = SocketFactory.getDefault().createSocket(this.host, this.port); + this.setSocketAttributes(socket); + NetSocketWriter writer; + if (messageFormat == MessageFormats.FORMAT_CUSTOM){ + Constructor ctor = customSocketWriter.getConstructor(Socket.class); + writer = BeanUtils.instantiateClass(ctor, socket); + } else { + writer = new NetSocketWriter(socket); + } + writer.setMessageFormat(messageFormat); + this.writer = writer; + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + return this.writer; + } + + /** + * @param customSocketWriter the customSocketWriter to set + * @throws ClassNotFoundException + */ + @SuppressWarnings("unchecked") + public void setCustomSocketWriteriClassName( + String customSocketWriterClassName) throws ClassNotFoundException { + this.customSocketWriter = (Class) Class + .forName(customSocketWriterClassName); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java index ef56331970..cd5646a5a6 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -17,6 +17,7 @@ package org.springframework.integration.ip.tcp; import java.io.IOException; +import java.lang.reflect.Constructor; import java.net.InetSocketAddress; import java.net.Socket; import java.net.SocketException; @@ -28,12 +29,14 @@ import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; +import org.springframework.beans.BeanUtils; 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. + * be required with large numbers of connections and incoming traffic. The + * number of threads is controlled by the poolSize property. * * @author Gary Russell * @@ -42,6 +45,8 @@ public class TcpNioReceivingChannelAdapter extends AbstractTcpReceivingChannelAdapter { protected ServerSocketChannel serverChannel; + protected boolean usingDirectBuffers; + protected Class customSocketReader; /** * Constructs a TcpNioReceivingChannelAdapter to listen on the port. @@ -152,8 +157,9 @@ public class TcpNioReceivingChannelAdapter extends SocketChannel channel = (SocketChannel) key.channel(); if (messageFormat == MessageFormats.FORMAT_CUSTOM) { try { - reader = (NioSocketReader) customSocketReader.newInstance(); - reader.setChannel(channel); + Constructor ctor = customSocketReader + .getConstructor(SocketChannel.class); + reader = BeanUtils.instantiateClass(ctor, channel); } catch (Exception e) { // TODO Auto-generated catch block e.printStackTrace(); @@ -184,16 +190,7 @@ public class TcpNioReceivingChannelAdapter extends 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(); - } - } + } catch (Exception e) {} } @Override @@ -207,5 +204,24 @@ public class TcpNioReceivingChannelAdapter extends } } + /** + * @param usingDirectBuffers Set true if you wish to use direct buffers + * for NIO operations. + */ + public void setUsingDirectBuffers(boolean usingDirectBuffers) { + this.usingDirectBuffers = usingDirectBuffers; + } + + /** + * @param customSocketReader the customSocketReader to set + * @throws ClassNotFoundException + */ + @SuppressWarnings("unchecked") + public void setCustomSocketReaderClassName(String customSocketReaderClassName) + throws ClassNotFoundException { + this.customSocketReader = (Class) Class + .forName(customSocketReaderClassName); + } + } \ No newline at end of file diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java new file mode 100644 index 0000000000..cb270d8ef1 --- /dev/null +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java @@ -0,0 +1,90 @@ +/* + * 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.lang.reflect.Constructor; +import java.nio.channels.SocketChannel; + +import org.springframework.beans.BeanUtils; + + +/** + * @author Gary Russell + * + */ +public class TcpNioSendingMessageHandler extends + AbstractTcpSendingMessageHandler { + + protected volatile SocketChannel socketChannel; + + protected boolean usingDirectBuffers; + + protected Class customSocketWriter; + + /** + * @param host + * @param port + */ + public TcpNioSendingMessageHandler(String host, int port) { + super(host, port); + } + + /** + * @return the socket + */ + protected synchronized SocketWriter getWriter() { + if (socketChannel == null) { + try { + socketChannel = SocketChannel.open(this.destinationAddress); + this.setSocketAttributes(socketChannel.socket()); + NioSocketWriter writer; + if (messageFormat == MessageFormats.FORMAT_CUSTOM){ + Constructor ctor = customSocketWriter.getConstructor(SocketChannel.class); + writer = BeanUtils.instantiateClass(ctor, socketChannel); + } else { + writer = new NioSocketWriter(socketChannel); + } + writer.setMessageFormat(messageFormat); + writer.setUsingDirectBuffers(usingDirectBuffers); + this.writer = writer; + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + 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 customSocketWriter the customSocketWriter to set + * @throws ClassNotFoundException + */ + @SuppressWarnings("unchecked") + public void setCustomSocketWriteriClassName( + String customSocketWriterClassName) throws ClassNotFoundException { + this.customSocketWriter = (Class) Class + .forName(customSocketWriterClassName); + } + +} diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java index beb98dbef8..2f856fffb8 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java @@ -110,7 +110,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler this.timeToLive = timeToLive; } - protected DatagramSocket getSocket() throws IOException { + protected synchronized DatagramSocket getSocket() throws IOException { if (this.socket == null) { MulticastSocket socket = new MulticastSocket(); if (this.timeToLive >= 0) { diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index 396dca015f..dd8ef3d85d 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -25,7 +25,6 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; @@ -74,8 +73,6 @@ public class UnicastSendingMessageHandler extends protected volatile DatagramSocket ackSocket; - protected volatile ExecutorService executorService; - protected volatile Exception fatalException; @@ -215,7 +212,7 @@ public class UnicastSendingMessageHandler extends socket.send(packet); } - protected DatagramSocket getSocket() throws IOException { + protected synchronized DatagramSocket getSocket() throws IOException { if (this.socket == null) { this.socket = new DatagramSocket(); setSocketAttributes(this.socket); diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java new file mode 100644 index 0000000000..5269b618a6 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.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; + +/** + * Writes packets that are always 24 bytes long. + * @author Gary Russell + * + */ +public class CustomNetSocketWriter extends NetSocketWriter { + + /** + * @param socket + */ + public CustomNetSocketWriter(Socket socket) { + super(socket); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.NetSocketWriter#writeCustomFormat(byte[]) + */ + @Override + protected void writeCustomFormat(byte[] bytes) throws IOException { + if (bytes.length > 24) { + socket.getOutputStream().write(bytes, 0, 24); + return; + } + socket.getOutputStream().write(bytes); + if (bytes.length < 24) { + socket.getOutputStream().write( + " ".substring(bytes.length) .getBytes()); + } + } + + + +} diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java new file mode 100644 index 0000000000..d56a362144 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java @@ -0,0 +1,57 @@ +/* + * 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; + +/** + * Writes packets that are always 24 bytes long. + * @author Gary Russell + * + */ +public class CustomNioSocketWriter extends NioSocketWriter { + + /** + * @param socket + */ + public CustomNioSocketWriter(SocketChannel channel) { + super(channel); + } + + /* (non-Javadoc) + * @see org.springframework.integration.ip.tcp.NetSocketWriter#writeCustomFormat(byte[]) + */ + @Override + protected void writeCustomFormat(byte[] bytes) throws IOException { + ByteBuffer data = ByteBuffer.wrap(bytes); + if (bytes.length > 24) { + data.limit(24); + channel.write(data); + return; + } + channel.write(data); + if (bytes.length < 24) { + data = ByteBuffer.wrap( + " ".substring(bytes.length).getBytes()); + channel.write(data); + } + } + + + +} 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 index f005fc42d5..32e2665504 100644 --- 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 @@ -37,7 +37,7 @@ public class NetSocketReaderTests { */ @Test public void testReadLength() throws Exception { - int port = 23556; + int port = Utils.findAvailableServerSocket(); ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Utils.testSendLength(port, null); Socket socket = server.accept(); @@ -57,6 +57,7 @@ public class NetSocketReaderTests { else { fail("Failed to assemble second message"); } + server.close(); } /** @@ -65,7 +66,7 @@ public class NetSocketReaderTests { */ @Test public void testReadStxEtx() throws Exception { - int port = 23557; + int port = Utils.findAvailableServerSocket(); ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Utils.testSendStxEtx(port, null); Socket socket = server.accept(); @@ -86,6 +87,7 @@ public class NetSocketReaderTests { else { fail("Failed to assemble second message"); } + server.close(); } /** @@ -94,7 +96,7 @@ public class NetSocketReaderTests { */ @Test public void testReadCrLf() throws Exception { - int port = 23558; + int port = Utils.findAvailableServerSocket(); ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); Utils.testSendCrLf(port, null); Socket socket = server.accept(); @@ -115,6 +117,7 @@ public class NetSocketReaderTests { else { fail("Failed to assemble second message"); } + server.close(); } } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java new file mode 100644 index 0000000000..4409c63547 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java @@ -0,0 +1,143 @@ +/* + * 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.*; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.ByteBuffer; + +import javax.net.ServerSocketFactory; +import javax.net.SocketFactory; + +import org.junit.Test; + +/** + * @author Gary Russell + * + */ +public class NetSocketWriterTests { + + @Test + public void testWriteLengthHeader() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + NetSocketWriter writer = new NetSocketWriter(socket); + writer.setMessageFormat(MessageFormats.FORMAT_LENGTH_HEADER); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 4]; + readFully(is, buff); + ByteBuffer buffer = ByteBuffer.wrap(buff); + assertEquals(testString.length(), buffer.getInt()); + assertEquals(testString, new String(buff, 4, testString.length())); + server.close(); + } + + @Test + public void testWriteStxEtx() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + NetSocketWriter writer = new NetSocketWriter(socket); + writer.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testWriteCrLf() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + NetSocketWriter writer = new NetSocketWriter(socket); + writer.setMessageFormat(MessageFormats.FORMAT_CRLF); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(testString, new String(buff, 0, testString.length())); + assertEquals('\r', buff[testString.length()]); + assertEquals('\n', buff[testString.length() + 1]); + server.close(); + } + + /** + * @param is + * @param buff + */ + private void readFully(InputStream is, byte[] buff) throws IOException { + for (int i = 0; i < buff.length; i++) { + buff[i] = (byte) is.read(); + } + } + +} diff --git a/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 index d7386263f9..7631a30bdc 100644 --- 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 @@ -44,7 +44,7 @@ public class NioSocketReaderTests { public void testReadLength() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = 23456; + int port = Utils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); @@ -93,13 +93,14 @@ public class NioSocketReaderTests { } } assertEquals("Did not receive data", 2, count); + server.close(); } @Test public void testFragmented() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = 23457; + int port = Utils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); @@ -148,6 +149,7 @@ public class NioSocketReaderTests { } } assertTrue("Did not receive data", done); + server.close(); } /** @@ -157,7 +159,7 @@ public class NioSocketReaderTests { public void testReadStxEtx() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = 23458; + int port = Utils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); @@ -207,6 +209,7 @@ public class NioSocketReaderTests { } } assertEquals("Did not receive data", 2, count); + server.close(); } /** @@ -216,7 +219,7 @@ public class NioSocketReaderTests { public void testReadCrLf() throws Exception { ServerSocketChannel server = ServerSocketChannel.open(); server.configureBlocking(false); - int port = 23459; + int port = Utils.findAvailableServerSocket(); server.socket().bind(new InetSocketAddress(port)); final Selector selector = Selector.open(); server.register(selector, SelectionKey.OP_ACCEPT); @@ -266,6 +269,7 @@ public class NioSocketReaderTests { } } assertEquals("Did not receive data", 2, count); + server.close(); } } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java new file mode 100644 index 0000000000..67784be253 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java @@ -0,0 +1,244 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.assertEquals; + +import java.io.IOException; +import java.io.InputStream; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; + +import javax.net.ServerSocketFactory; + +import org.junit.Test; + +/** + * @author Gary Russell + * + */ +public class NioSocketWriterTests { + + @Test + public void testWriteLengthHeader() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost", port)); + NioSocketWriter writer = new NioSocketWriter(channel); + writer.setMessageFormat(MessageFormats.FORMAT_LENGTH_HEADER); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 4]; + readFully(is, buff); + ByteBuffer buffer = ByteBuffer.wrap(buff); + assertEquals(testString.length(), buffer.getInt()); + assertEquals(testString, new String(buff, 4, testString.length())); + server.close(); + } + + @Test + public void testWriteStxEtx() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost", port)); + NioSocketWriter writer = new NioSocketWriter(channel); + writer.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testWriteCrLf() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost", port)); + NioSocketWriter writer = new NioSocketWriter(channel); + writer.setMessageFormat(MessageFormats.FORMAT_CRLF); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(testString, new String(buff, 0, testString.length())); + assertEquals('\r', buff[testString.length()]); + assertEquals('\n', buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testWriteLengthHeaderDirect() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost", port)); + NioSocketWriter writer = new NioSocketWriter(channel); + writer.setMessageFormat(MessageFormats.FORMAT_LENGTH_HEADER); + writer.setUsingDirectBuffers(true); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 4]; + readFully(is, buff); + ByteBuffer buffer = ByteBuffer.wrap(buff); + assertEquals(testString.length(), buffer.getInt()); + assertEquals(testString, new String(buff, 4, testString.length())); + server.close(); + } + + @Test + public void testWriteStxEtxDirect() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost", port)); + NioSocketWriter writer = new NioSocketWriter(channel); + writer.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + writer.setUsingDirectBuffers(true); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testWriteCrLfDirect() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + ByteBuffer buffer = ByteBuffer.allocate(testString.length()); + buffer.put(testString.getBytes()); + SocketChannel channel = SocketChannel.open(new InetSocketAddress("localhost", port)); + NioSocketWriter writer = new NioSocketWriter(channel); + writer.setMessageFormat(MessageFormats.FORMAT_CRLF); + writer.setUsingDirectBuffers(true); + writer.write(buffer.array()); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(testString, new String(buff, 0, testString.length())); + assertEquals('\r', buff[testString.length()]); + assertEquals('\n', buff[testString.length() + 1]); + server.close(); + } + + + /** + * @param is + * @param buff + */ + private void readFully(InputStream is, byte[] buff) throws IOException { + for (int i = 0; i < buff.length; i++) { + buff[i] = (byte) is.read(); + } + } + + } diff --git a/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 index 9844a151c2..fcd0b03da0 100644 --- 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 @@ -21,10 +21,10 @@ 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; +import org.springframework.integration.message.MessageBuilder; /** * @author Gary Russell @@ -54,11 +54,16 @@ public class SocketMessageMapperTests { /** * Test method for {@link org.springframework.integration.ip.tcp.SocketMessageMapper#fromMessage(org.springframework.integration.core.Message)}. + * @throws Exception */ @Test - @Ignore - public void testFromMessage() { - fail("Not yet implemented"); + public void testFromMessage() throws Exception { + String s = "test"; + Message message = MessageBuilder.withPayload(s).build(); + SocketMessageMapper mapper = new SocketMessageMapper(); + byte[] bArray = mapper.fromMessage(message); + assertEquals(s, new String(bArray)); + } @@ -90,9 +95,4 @@ public class SocketMessageMapperTests { } - 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 index 43920a9783..e333992751 100644 --- 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 @@ -21,6 +21,7 @@ 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.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** @@ -35,8 +36,8 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNet() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = 12345; - AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); + int port = Utils.findAvailableServerSocket(); + AbstractInternetProtocolReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); adapter.setOutputChannel(channel); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.initialize(); @@ -53,6 +54,7 @@ public class TcpReceivingChannelAdapterTests { assertNotNull(message); assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String((byte[])message.getPayload())); + adapter.stop(); } /** @@ -62,8 +64,8 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNetCustom() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = 12346; - AbstractTcpReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); + int port = Utils.findAvailableServerSocket(); + TcpNetReceivingChannelAdapter adapter = new TcpNetReceivingChannelAdapter(port); adapter.setOutputChannel(channel); adapter.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNetSocketReader"); adapter.setMessageFormat(MessageFormats.FORMAT_CUSTOM); @@ -82,6 +84,7 @@ public class TcpReceivingChannelAdapterTests { assertNotNull(message); assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); + adapter.stop(); } @@ -91,7 +94,7 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNio() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = 12355; + int port = Utils.findAvailableServerSocket(); TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); adapter.setOutputChannel(channel); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); @@ -109,6 +112,7 @@ public class TcpReceivingChannelAdapterTests { assertNotNull(message); assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String((byte[])message.getPayload())); + adapter.stop(); } /** @@ -117,7 +121,7 @@ public class TcpReceivingChannelAdapterTests { @Test public void testNioCustom() throws Exception { QueueChannel channel = new QueueChannel(2); - int port = 12356; + int port = Utils.findAvailableServerSocket(); TcpNioReceivingChannelAdapter adapter = new TcpNioReceivingChannelAdapter(port); adapter.setOutputChannel(channel); adapter.setCustomSocketReaderClassName("org.springframework.integration.ip.tcp.CustomNioSocketReader"); @@ -137,6 +141,7 @@ public class TcpReceivingChannelAdapterTests { assertNotNull(message); assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", new String((byte[])message.getPayload())); + adapter.stop(); } } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java new file mode 100644 index 0000000000..8b3dd5e1af --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -0,0 +1,298 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.ip.tcp; + +import static org.junit.Assert.assertEquals; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; + +import javax.net.ServerSocketFactory; + +import org.junit.Test; +import org.springframework.integration.core.Message; +import org.springframework.integration.message.MessageBuilder; + + +/** + * @author Gary Russell + * + */ +public class TcpSendingMessageHandlerTests { + + @Test + public void testNetBlocking() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNetSendingMessageHandler handler = new TcpNetSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + handler.setBlockingWrite(true); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testNetNonBlocking() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNetSendingMessageHandler handler = new TcpNetSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + handler.setBlockingWrite(false); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testNetCustom() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNetSendingMessageHandler handler = new TcpNetSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_CUSTOM); + handler.setBlockingWrite(true); + handler.setCustomSocketWriteriClassName("org.springframework.integration.ip.tcp.CustomNetSocketWriter"); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[24]; + readFully(is, buff); + assertEquals((testString + " ").substring(0, 24), + new String(buff)); + server.close(); + } + + + @Test + public void testNioBlocking() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + handler.setBlockingWrite(true); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testNioNonBlocking() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + handler.setBlockingWrite(false); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testNioBlockingDirect() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + handler.setBlockingWrite(true); + handler.setUsingDirectBuffers(true); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testNioNonBlockingDirect() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_STX_ETX); + handler.setBlockingWrite(false); + handler.setUsingDirectBuffers(true); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[testString.length() + 2]; + readFully(is, buff); + assertEquals(MessageFormats.STX, buff[0]); + assertEquals(testString, new String(buff, 1, testString.length())); + assertEquals(MessageFormats.ETX, buff[testString.length() + 1]); + server.close(); + } + + @Test + public void testNioCustom() throws Exception { + final int port = Utils.findAvailableServerSocket(); + final String testString = "abcdef"; + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + Thread t = new Thread(new Runnable() { + public void run() { + try { + TcpNioSendingMessageHandler handler = new TcpNioSendingMessageHandler("localhost", port); + handler.setMessageFormat(MessageFormats.FORMAT_CUSTOM); + handler.setBlockingWrite(true); + handler.setCustomSocketWriteriClassName("org.springframework.integration.ip.tcp.CustomNioSocketWriter"); + Message message = MessageBuilder.withPayload(testString).build(); + handler.handleMessage(message); + Thread.sleep(1000000000L); + } catch (Exception e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + Socket socket = server.accept(); + InputStream is = socket.getInputStream(); + byte[] buff = new byte[24]; + readFully(is, buff); + assertEquals((testString + " ").substring(0, 24), + new String(buff)); + server.close(); + } + + /** + * @param is + * @param buff + */ + private void readFully(InputStream is, byte[] buff) throws IOException { + for (int i = 0; i < buff.length; i++) { + buff[i] = (byte) is.read(); + } + } +} diff --git a/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 index 7075a5fbba..85fc214b60 100644 --- 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 @@ -17,10 +17,13 @@ package org.springframework.integration.ip.tcp; import java.io.OutputStream; import java.net.InetAddress; +import java.net.ServerSocket; import java.net.Socket; import java.nio.ByteBuffer; import java.util.concurrent.CountDownLatch; +import javax.net.ServerSocketFactory; + /** * TCP/IP Test utilities. * @@ -164,4 +167,15 @@ public class Utils { thread.setDaemon(true); thread.start(); } + + public static int findAvailableServerSocket() { + for (int i = 5678; i < 5878; i++) { + try { + ServerSocket sock = ServerSocketFactory.getDefault().createServerSocket(i); + sock.close(); + return i; + } catch (Exception e) { } + } + throw new RuntimeException("Cannot find a free server socket"); + } }