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

Merge pull request #762 from garyrussell/INT-2956

Browse Source 
* INT-2956: Fix NIO With Custom Deserializers
This commit is contained in:
Mark Fisher
2013-05-07 12:39:07 -04:00
parent 3eab287e95 474a00509b
commit 121b28f426
2 changed files with 132 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -547,6 +547,33 @@ public class TcpNioConnection extends TcpConnectionSupport {
private volatile boolean isClosed;
@Override
public int read(byte[] b, int off, int len) throws IOException {
Assert.notNull(b, "byte[] cannot be null");
if (off < 0 || len < 0 || len > b.length - off) {
throw new IndexOutOfBoundsException();
}
else if (len == 0) {
return 0;
}
int n = 0;
while ((this.available.get() > 0 || n == 0) &&
n < len) {
int bite = read();
if (bite < 0) {
if (n == 0) {
return -1;
}
else {
return n;
}
}
b[off + n++] = (byte) bite;
}
return n;
}
@Override
public synchronized int read() throws IOException {
if (this.isClosed && available.get() == 0) {

105
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java
View File

@@ -53,7 +53,9 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.Message;
import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.SocketUtils;
@@ -337,6 +339,109 @@ public class TcpNioConnectionTests {
assertTrue(messageLatch.await(10, TimeUnit.SECONDS));
}
@Test
public void testByteArrayRead() throws Exception {
SocketChannel socketChannel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
when(socketChannel.socket()).thenReturn(socket);
TcpNioConnection connection = new TcpNioConnection(socketChannel, false, false, null, null);
TcpNioConnection.ChannelInputStream stream = (ChannelInputStream) new DirectFieldAccessor(connection)
.getPropertyValue("channelInputStream");
stream.write("foo".getBytes(), 3);
byte[] out = new byte[2];
int n = stream.read(out);
assertEquals(2, n);
assertEquals("fo", new String(out));
out = new byte[2];
n = stream.read(out);
assertEquals(1, n);
assertEquals("o\u0000", new String(out));
}
@Test
public void testByteArrayReadMulti() throws Exception {
SocketChannel socketChannel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
when(socketChannel.socket()).thenReturn(socket);
TcpNioConnection connection = new TcpNioConnection(socketChannel, false, false, null, null);
TcpNioConnection.ChannelInputStream stream = (ChannelInputStream) new DirectFieldAccessor(connection)
.getPropertyValue("channelInputStream");
stream.write("foo".getBytes(), 3);
stream.write("bar".getBytes(), 3);
byte[] out = new byte[6];
int n = stream.read(out);
assertEquals(6, n);
assertEquals("foobar", new String(out));
}
@Test
public void testByteArrayReadWithOffset() throws Exception {
SocketChannel socketChannel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
when(socketChannel.socket()).thenReturn(socket);
TcpNioConnection connection = new TcpNioConnection(socketChannel, false, false, null, null);
TcpNioConnection.ChannelInputStream stream = (ChannelInputStream) new DirectFieldAccessor(connection)
.getPropertyValue("channelInputStream");
stream.write("foo".getBytes(), 3);
byte[] out = new byte[5];
int n = stream.read(out, 1, 4);
assertEquals(3, n);
assertEquals("\u0000foo\u0000", new String(out));
}
@Test
public void testByteArrayReadWithBadArgs() throws Exception {
SocketChannel socketChannel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
when(socketChannel.socket()).thenReturn(socket);
TcpNioConnection connection = new TcpNioConnection(socketChannel, false, false, null, null);
TcpNioConnection.ChannelInputStream stream = (ChannelInputStream) new DirectFieldAccessor(connection)
.getPropertyValue("channelInputStream");
stream.write("foo".getBytes(), 3);
byte[] out = new byte[5];
try {
stream.read(out, 1, 5);
fail("Expected IndexOutOfBoundsException");
}
catch (IndexOutOfBoundsException e) {}
try {
stream.read(null, 1, 5);
fail("Expected IllegalArgumentException");
}
catch (IllegalArgumentException e) {}
assertEquals(0, stream.read(out, 0, 0));
assertEquals(3, stream.read(out));
}
@Test
public void testByteArrayBlocksForZeroRead() throws Exception {
SocketChannel socketChannel = mock(SocketChannel.class);
Socket socket = mock(Socket.class);
when(socketChannel.socket()).thenReturn(socket);
TcpNioConnection connection = new TcpNioConnection(socketChannel, false, false, null, null);
final TcpNioConnection.ChannelInputStream stream = (ChannelInputStream) new DirectFieldAccessor(connection)
.getPropertyValue("channelInputStream");
final CountDownLatch latch = new CountDownLatch(1);
final byte[] out = new byte[4];
ExecutorService exec = Executors.newSingleThreadExecutor();
exec.execute(new Runnable(){
public void run() {
try {
stream.read(out);
}
catch (IOException e) {
e.printStackTrace();
}
latch.countDown();
}
});
Thread.sleep(1000);
assertEquals(0x00, out[0]);
stream.write("foo".getBytes(), 3);
assertTrue(latch.await(10, TimeUnit.SECONDS));
assertEquals("foo\u0000", new String(out));
}
private void readFully(InputStream is, byte[] buff) throws IOException {
for (int i = 0; i < buff.length; i++) {
buff[i] = (byte) is.read();