From db26f7475099567c4ee73185181458dc9d6445b6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 16 Apr 2010 23:17:31 +0000 Subject: [PATCH] tcp/ip - NIO - improve direct buffer management; fix thread safety for non-direct buffers. --- .../integration/ip/tcp/NioSocketWriter.java | 187 +++++-- .../ip/tcp/TcpNioSendingMessageHandler.java | 21 +- .../ip/tcp/CustomNioSocketWriter.java | 10 +- .../ip/tcp/NioSocketWriterTests.java | 496 ++++++++++-------- 4 files changed, 453 insertions(+), 261 deletions(-) diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java index 03b9594aea..b3743dcb85 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketWriter.java @@ -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 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(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) diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java index f84a6f712f..6b1471caaf 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioSendingMessageHandler.java @@ -33,6 +33,8 @@ public class TcpNioSendingMessageHandler extends protected boolean usingDirectBuffers; protected Class 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 ctor = customSocketWriter.getConstructor(SocketChannel.class); - writer = BeanUtils.instantiateClass(ctor, socketChannel); + Constructor 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; + } + } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java index d56a362144..3baf8f3ab9 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/CustomNioSocketWriter.java @@ -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) diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java index f888647dc3..6701d8ce13 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/NioSocketWriterTests.java @@ -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(); + } + } + +}