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

INT-781 Inbound TCP Adapters & Tests - no namespace support yet.

Browse Source
This commit is contained in:
Gary Russell
2010-02-25 00:59:13 +00:00
parent 98243c32c6
commit 6e2295b625
27 changed files with 1772 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java
View File

@@ -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;
}

34
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketReader.java
View File

@@ -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,
*/

110
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractSocketWriter.java Normal file
View File

@@ -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;
}
}

22
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java
View File

@@ -43,10 +43,6 @@ public abstract class AbstractTcpReceivingChannelAdapter extends
protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER;
protected Class<SocketReader> 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<SocketReader>) Class.forName(customSocketReaderClassName);
}
/**
* @param usingDirectBuffers the usingDirectBuffers to set
*/
public void setUsingDirectBuffers(boolean usingDirectBuffers) {
this.usingDirectBuffers = usingDirectBuffers;
}
/**
* @param poolSize the poolSize to set
*/

190
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java Normal file
View File

@@ -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;
}
}

21
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketReader.java
View File

@@ -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;
}
}

91
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NetSocketWriter.java Normal file
View File

@@ -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) {}
}
}

87
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java
View File

@@ -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;
}
}

174
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java Normal file
View File

@@ -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) {}
}
}

47
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketMessageMapper.java
View File

@@ -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<SocketReader>,
OutboundMessageMapper<SocketWriter> {
OutboundMessageMapper<byte[]> {
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;
}
}

11
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/SocketWriter.java
View File

@@ -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;
}

30
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetReceivingChannelAdapter.java
View File

@@ -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<NetSocketReader> 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<NetSocketReader> 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<NetSocketReader>) Class
.forName(customSocketReaderClassName);
}
}

85
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNetSendingMessageHandler.java Normal file
View File

@@ -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<NetSocketWriter> 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<NetSocketWriter> 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<NetSocketWriter>) Class
.forName(customSocketWriterClassName);
}
}

42
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java
View File

@@ -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<NioSocketReader> 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<NioSocketReader> 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<NioSocketReader>) Class
.forName(customSocketReaderClassName);
}
}

90
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java Normal file
View File

@@ -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<NioSocketWriter> 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<NioSocketWriter> 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<NioSocketWriter>) Class
.forName(customSocketWriterClassName);
}
}

2
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java
View File

@@ -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) {

5
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -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);

53
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNetSocketWriter.java Normal file
View File

@@ -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());
}
}
}

57
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java Normal file
View File

@@ -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);
}
}
}

9
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketReaderTests.java
View File

@@ -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();
}
}

143
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NetSocketWriterTests.java Normal file
View File

@@ -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();
}
}
}

12
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketReaderTests.java
View File

@@ -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();
}
}

244
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java Normal file
View File

@@ -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();
}
}
}

18
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/SocketMessageMapperTests.java
View File

@@ -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<String> 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 {
}
}

17
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

@@ -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();
}
}

298
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java Normal file
View File

@@ -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<String> 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<String> 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<String> 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<String> 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<String> 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<String> 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<String> 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<String> 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();
}
}
}

14
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java
View File

@@ -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");
}
}