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[]`.

Conflicts:
	spring-integration-ip/src/test/java/org/springframework/integration/ip/util/SocketTestUtils.java

* Change Lambda for `EventPublisher` to `Mock`
This commit is contained in:
Gary Russell
2016-09-26 17:53:23 -04:00
committed by Artem Bilan
parent aafeb3a55f
commit 166a732447
4 changed files with 18 additions and 12 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

@@ -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();

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;
@@ -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());
}

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

@@ -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);

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

@@ -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<Void>() {