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

INT-4124: ByteBuffer.array() with Direct Buffers

Browse Source 
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[]`.
This commit is contained in:
Gary Russell
2016-09-26 17:53:23 -04:00
committed by Artem Bilan
parent e5ae7d886d
commit 93989a0c9b
5 changed files with 17 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -441,7 +441,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();
}
@@ -723,10 +723,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();

3
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNetConnectionTests.java
View File

@@ -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;
@@ -100,7 +101,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());
}

4
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -54,7 +54,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(e -> { });
scf.setSerializer(serializer);
scf.setDeserializer(serializer);
scf.registerListener(listener);

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

@@ -386,7 +386,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);
@@ -405,8 +405,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);
@@ -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[5];
int n = stream.read(out, 1, 4);
assertEquals(3, n);
@@ -436,7 +436,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);
@@ -480,7 +480,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));
}
@@ -760,7 +760,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<Void>() {

2
spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java
View File

@@ -234,7 +234,7 @@ public class SocketTestUtils {
testCompleteLatch.await(10, TimeUnit.SECONDS);
}
catch (Exception e1) {
e1.printStackTrace();
logger.debug("write failed", e1);
}
finally {
if (socket != null) {