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

tcp/ip - NIO - improve direct buffer management; fix thread safety for non-direct buffers.

Browse Source
This commit is contained in:
Gary Russell
2010-04-16 23:17:31 +00:00
parent 87d994cd6d
commit db26f74750
4 changed files with 453 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java
View File

@@ -18,6 +18,8 @@ package org.springframework.integration.ip.tcp;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
/**
* A {@link SocketWriter} that writes to a {@link java.nio.channels.SocketChannel}. The
@@ -60,11 +62,34 @@ public class NioSocketWriter extends AbstractSocketWriter {
*/
protected ByteBuffer crLfPart;
/**
* If we are using direct buffers, we don't want to churn them using
* normal heap management. But,
* because we can have multiple threads writing and we might write in
* chunks, we need a dedicated buffer for each thread; up to a limit.
* We handle this with a blocking queue.
*/
protected BlockingQueue<ByteBuffer> buffers;
protected int maxBuffers = 2;
protected int bufferCount = 0;
private int sendBufferSize;
/**
* @param socket
*/
public NioSocketWriter(SocketChannel channel) {
public NioSocketWriter(SocketChannel channel,
int maxBuffers,
int sendBufferSize) {
this.channel = channel;
this.maxBuffers = maxBuffers;
if (sendBufferSize <= 0) {
sendBufferSize = 2048;
}
this.sendBufferSize = sendBufferSize;
buffers = new LinkedBlockingQueue<ByteBuffer>(maxBuffers);
}
/**
@@ -74,6 +99,33 @@ public class NioSocketWriter extends AbstractSocketWriter {
this.usingDirectBuffers = usingDirectBuffers;
}
protected ByteBuffer getBuffer() throws InterruptedException {
ByteBuffer buffer = this.buffers.poll();
if (buffer != null) {
return buffer;
}
synchronized (buffers) {
if (bufferCount < maxBuffers) {
bufferCount++;
return ByteBuffer.allocateDirect(this.sendBufferSize);
}
// another thread may have returned one while we were sync'd
buffer = this.buffers.poll();
if (buffer != null) {
return buffer;
}
}
buffer = this.buffers.take();
buffer.clear();
return buffer;
}
protected void returnBuffer(ByteBuffer buffer) {
if (buffer != null) {
buffers.offer(buffer);
}
}
/* (non-Javadoc)
* @see org.springframework.integration.ip.tcp.AbstractSocketWriter#writeCrLfFormat(byte[])
*/
@@ -81,22 +133,31 @@ public class NioSocketWriter extends AbstractSocketWriter {
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;
try {
checkBufferSize(bytes, 2);
buffer = getBuffer();
buffer.put(bytes);
buffer.put((byte) '\r');
buffer.put((byte) '\n');
buffer.flip();
channel.write(buffer);
return;
} catch (InterruptedException e) {
throw new IOException("Could not get buffer", e);
} finally {
returnBuffer(buffer);
}
}
if (crLfPart == null) {
crLfPart = ByteBuffer.allocate(2);
crLfPart.put((byte) '\r');
crLfPart.put((byte) '\n');
synchronized (channel) {
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);
}
channel.write(ByteBuffer.wrap(bytes));
crLfPart.flip();
channel.write(crLfPart);
}
/* (non-Javadoc)
@@ -113,23 +174,31 @@ public class NioSocketWriter extends AbstractSocketWriter {
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;
try {
checkBufferSize(bytes, 4);
buffer = getBuffer();
buffer.putInt(bytes.length);
buffer.put(bytes);
buffer.flip();
channel.write(buffer);
return;
} catch (InterruptedException e) {
throw new IOException("Could not get buffer", e);
} finally {
returnBuffer(buffer);
}
}
if (lengthPart == null) {
lengthPart = ByteBuffer.allocate(4);
} else {
lengthPart.clear();
}
lengthPart.putInt(bytes.length);
lengthPart.flip();
channel.write(lengthPart);
channel.write(ByteBuffer.wrap(bytes));
synchronized (channel) {
if (lengthPart == null) {
lengthPart = ByteBuffer.allocate(4);
} else {
lengthPart.clear();
}
lengthPart.putInt(bytes.length);
lengthPart.flip();
channel.write(lengthPart);
channel.write(ByteBuffer.wrap(bytes));
}
}
/* (non-Javadoc)
@@ -139,25 +208,47 @@ public class NioSocketWriter extends AbstractSocketWriter {
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;
try {
checkBufferSize(bytes, 2);
buffer = getBuffer();
buffer.put((byte) STX);
buffer.put(bytes);
buffer.put((byte) ETX);
buffer.flip();
channel.write(buffer);
return;
} catch (InterruptedException e) {
throw new IOException("Could not get buffer", e);
} finally {
returnBuffer(buffer);
}
}
if (stxPart == null) {
stxPart = ByteBuffer.allocate(1);
stxPart.put((byte) STX);
etxPart = ByteBuffer.allocate(1);
etxPart.put((byte) ETX);
synchronized (channel) {
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);
}
}
/**
* @param bytes
* @throws IOException
*/
private void checkBufferSize(byte[] bytes, int pad) throws IOException {
if (bytes.length + pad > sendBufferSize) {
throw new IOException("Send buffer too small (" + sendBufferSize +
") increase so-send-buffer-size to at least " +
bytes.length + pad);
}
stxPart.flip();
channel.write(stxPart);
channel.write(ByteBuffer.wrap(bytes));
etxPart.flip();
channel.write(etxPart);
}
/* (non-Javadoc)

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

@@ -33,6 +33,8 @@ public class TcpNioSendingMessageHandler extends
protected boolean usingDirectBuffers;
protected Class<NioSocketWriter> customSocketWriter;
protected int buffsPerConnection = 5;
/**
* @param host
@@ -52,10 +54,11 @@ public class TcpNioSendingMessageHandler extends
this.setSocketAttributes(socketChannel.socket());
NioSocketWriter writer;
if (messageFormat == MessageFormats.FORMAT_CUSTOM){
Constructor<NioSocketWriter> ctor = customSocketWriter.getConstructor(SocketChannel.class);
writer = BeanUtils.instantiateClass(ctor, socketChannel);
Constructor<NioSocketWriter> ctor = customSocketWriter
.getConstructor(SocketChannel.class, int.class, int.class);
writer = BeanUtils.instantiateClass(ctor, socketChannel, buffsPerConnection, soSendBufferSize);
} else {
writer = new NioSocketWriter(socketChannel);
writer = new NioSocketWriter(socketChannel, buffsPerConnection, soSendBufferSize);
}
writer.setMessageFormat(messageFormat);
writer.setUsingDirectBuffers(usingDirectBuffers);
@@ -86,4 +89,16 @@ public class TcpNioSendingMessageHandler extends
.forName(customSocketWriterClassName);
}
/**
* If direct buffers are being used, sets the max number of
* buffers allowed per connection. Defaults to 5. It is unlikely
* this would ever need to be changed. Each buffer is set at the
* soSendBufferSize or, if not set, 2048 bytes.
*
* @param buffsPerConnection the buffsPerConnection to set
*/
public void setBuffsPerConnection(int buffsPerConnection) {
this.buffsPerConnection = buffsPerConnection;
}
}

10
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java
View File

@@ -26,11 +26,15 @@ import java.nio.channels.SocketChannel;
*/
public class CustomNioSocketWriter extends NioSocketWriter {
/**
* @param socket
* @param channel
* @param maxBuffers
* @param sendBufferSize
*/
public CustomNioSocketWriter(SocketChannel channel) {
super(channel);
public CustomNioSocketWriter(SocketChannel channel, int maxBuffers,
int sendBufferSize) {
super(channel, maxBuffers, sendBufferSize);
}
/* (non-Javadoc)

496
org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java
View File

@@ -24,6 +24,7 @@ import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.CountDownLatch;
import javax.net.ServerSocketFactory;
@@ -32,214 +33,295 @@ import org.springframework.integration.ip.util.SocketUtils;
/**
* @author Gary Russell
*
*
*/
public class NioSocketWriterTests {
@Test
public void testWriteLengthHeader() throws Exception {
final int port = SocketUtils.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 = SocketUtils.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 = SocketUtils.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 = SocketUtils.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 = SocketUtils.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 = SocketUtils.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();
}
}
@Test
public void testBuffersNoWait() throws Exception {
NioSocketWriter writer = new NioSocketWriter(null, 2, 2048);
ByteBuffer b1 = writer.getBuffer();
ByteBuffer b2 = writer.getBuffer();
writer.returnBuffer(b2);
ByteBuffer b3 = writer.getBuffer();
assertEquals(b2, b3);
writer.returnBuffer(b3);
writer.returnBuffer(b1);
b3 = writer.getBuffer();
assertEquals(b2, b3);
writer.returnBuffer(b3);
b3 = writer.getBuffer();
assertEquals(b1, b3);
writer.returnBuffer(b3);
}
@Test
public void testBuffersWait() throws Exception {
final NioSocketWriter writer = new NioSocketWriter(null, 2, 2048);
ByteBuffer b1 = writer.getBuffer();
ByteBuffer b2 = writer.getBuffer();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
final ByteBuffer b2a = b2;
new Thread(new Runnable(){
public void run() {
latch1.countDown();
try {
ByteBuffer b = writer.getBuffer();
assertEquals(b2a, b);
writer.returnBuffer(b);
latch2.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}).start();
latch1.await();
Thread.sleep(2000);
writer.returnBuffer(b2);
latch2.await();
ByteBuffer b3 = writer.getBuffer();
assertEquals(b2, b3);
writer.returnBuffer(b3);
writer.returnBuffer(b1);
b3 = writer.getBuffer();
assertEquals(b2, b3);
writer.returnBuffer(b3);
b3 = writer.getBuffer();
assertEquals(b1, b3);
writer.returnBuffer(b3);
}
@Test
public void testWriteLengthHeader() throws Exception {
final int port = SocketUtils.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, 2,
2048);
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 = SocketUtils.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, 2,
2048);
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 = SocketUtils.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, 2,
2048);
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 = SocketUtils.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, 2,
2048);
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 = SocketUtils.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, 2,
2048);
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 = SocketUtils.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, 2,
2048);
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();
}
}
}