From 166a7324475e2af7fd3e1edd82197f96644528e4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 26 Sep 2016 17:53:23 -0400 Subject: [PATCH] INT-4124: ByteBuffer.array() with Direct Buffers JIRA: https://jira.spring.io/browse/INT-4124 `ByteBuffer.array()` returns null when Direct. Change `ChannelInputStream` to use the `ByteBuffer` directly instead of the underlying `byte[]`. Conflicts: spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java * Change Lambda for `EventPublisher` to `Mock` --- .../ip/tcp/connection/TcpNioConnection.java | 7 ++++--- .../ip/tcp/connection/TcpNetConnectionTests.java | 3 ++- .../tcp/connection/TcpNioConnectionReadTests.java | 6 +++++- .../ip/tcp/connection/TcpNioConnectionTests.java | 14 +++++++------- 4 files changed, 18 insertions(+), 12 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index bb82895546..01138cdcc2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -439,7 +439,7 @@ public class TcpNioConnection extends TcpConnectionSupport { if (logger.isTraceEnabled()) { logger.trace(this.getConnectionId() + " Sending " + rawBuffer.limit() + " to pipe"); } - this.channelInputStream.write(rawBuffer.array(), rawBuffer.limit()); + this.channelInputStream.write(rawBuffer); rawBuffer.clear(); } @@ -721,10 +721,11 @@ public class TcpNioConnection extends TcpConnectionSupport { * @param bytesToWrite * @throws IOException */ - public void write(byte[] array, int bytesToWrite) throws IOException { + public void write(ByteBuffer byteBuffer) throws IOException { + int bytesToWrite = byteBuffer.limit() - byteBuffer.position(); if (bytesToWrite > 0) { byte[] buffer = new byte[bytesToWrite]; - System.arraycopy(array, 0, buffer, 0, bytesToWrite); + byteBuffer.get(buffer); this.available.addAndGet(bytesToWrite); if (TcpNioConnection.this.writingLatch != null) { TcpNioConnection.this.writingLatch.countDown(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java index fc0c798338..18cdd6c613 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java @@ -27,6 +27,7 @@ import java.io.InputStream; import java.io.PipedInputStream; import java.io.PipedOutputStream; import java.net.Socket; +import java.nio.ByteBuffer; import java.nio.channels.SocketChannel; import java.util.concurrent.atomic.AtomicReference; @@ -104,7 +105,7 @@ public class TcpNetConnectionTests { TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, nullPublisher, null); ChannelInputStream inputStream = TestUtils.getPropertyValue(connection, "channelInputStream", ChannelInputStream.class); - inputStream.write(new byte[] {(byte) 0x80}, 1); + inputStream.write(ByteBuffer.wrap(new byte[] { (byte) 0x80 })); assertEquals(0x80, inputStream.read()); } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 3a6b9b4e4c..e85f9142e3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.ip.tcp.connection; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; import java.net.Socket; import java.util.ArrayList; @@ -31,6 +32,7 @@ import javax.net.SocketFactory; import org.junit.Test; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; import org.springframework.integration.ip.tcp.serializer.ByteArrayLengthHeaderSerializer; @@ -54,7 +56,9 @@ public class TcpNioConnectionReadTests { private AbstractServerConnectionFactory getConnectionFactory( AbstractByteArraySerializer serializer, TcpListener listener, TcpSender sender) throws Exception { - AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(0); + scf.setUsingDirectBuffers(true); + scf.setApplicationEventPublisher(mock(ApplicationEventPublisher.class)); scf.setSerializer(serializer); scf.setDeserializer(serializer); scf.registerListener(listener); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java index 59b57c6b8a..9636b29e2b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -421,7 +421,7 @@ public class TcpNioConnectionTests { 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(ByteBuffer.wrap("foo".getBytes())); byte[] out = new byte[2]; int n = stream.read(out); assertEquals(2, n); @@ -440,8 +440,8 @@ public class TcpNioConnectionTests { 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); + stream.write(ByteBuffer.wrap("foo".getBytes())); + stream.write(ByteBuffer.wrap("bar".getBytes())); byte[] out = new byte[6]; int n = stream.read(out); assertEquals(6, n); @@ -456,7 +456,7 @@ public class TcpNioConnectionTests { 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(ByteBuffer.wrap("foo".getBytes())); byte[] out = new byte[5]; int n = stream.read(out, 1, 4); assertEquals(3, n); @@ -471,7 +471,7 @@ public class TcpNioConnectionTests { 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(ByteBuffer.wrap("foo".getBytes())); byte[] out = new byte[5]; try { stream.read(out, 1, 5); @@ -512,7 +512,7 @@ public class TcpNioConnectionTests { }); Thread.sleep(1000); assertEquals(0x00, out[0]); - stream.write("foo".getBytes(), 3); + stream.write(ByteBuffer.wrap("foo".getBytes())); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertEquals("foo\u0000", new String(out)); } @@ -788,7 +788,7 @@ public class TcpNioConnectionTests { readerFinishedLatch.countDown(); return null; } - }).when(cis).write(any(byte[].class), Matchers.anyInt()); + }).when(cis).write(any(ByteBuffer.class)); doReturn(true).when(logger).isTraceEnabled(); doAnswer(new Answer() {