INT-2936 Fix TCP Binary DeSerialization with NIO

Inadvertent sign extension on binary data with bit 7 set
causes early termination of binary deserializers.

The ChannelInputStream (which replaced the piped input
and output streams) failed to mask off the top 24 bits
of "normal" bytes received, causing Deserializers to
believe the stream was closed.

Add a mask of 0xff to bytes read.

Add a test case.

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

@@ -68,7 +68,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 
 	private volatile long lastSend;
 
-	private AtomicInteger executionControl = new AtomicInteger();
+	private final AtomicInteger executionControl = new AtomicInteger();
 
 	private volatile boolean writingToPipe;
 
@@ -529,7 +529,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
 				}
 			}
 			int bite;
-			bite = this.currentBuffer[this.currentOffset++];
+			bite = this.currentBuffer[this.currentOffset++] & 0xff;
 			this.available.decrementAndGet();
 			if (this.currentOffset >= this.currentBuffer.length) {
 				this.currentBuffer = null;

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

@@ -23,6 +23,7 @@ import static org.mockito.Mockito.when;
 
 import java.io.InputStream;
 import java.net.Socket;
+import java.nio.channels.SocketChannel;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.Log;
@@ -31,7 +32,9 @@ import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.integration.ip.tcp.connection.TcpNioConnection.ChannelInputStream;
 import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer;
+import org.springframework.integration.test.util.TestUtils;
 
 /**
  * @author Gary Russell
@@ -68,4 +71,15 @@ public class TcpNetConnectionTests {
 				log.get());
 	}
 
+	@Test
+	public void testBinary() throws Exception {
+		SocketChannel socketChannel = mock(SocketChannel.class);
+		Socket socket = mock(Socket.class);
+		when(socketChannel.socket()).thenReturn(socket);
+		TcpNioConnection connection = new TcpNioConnection(socketChannel, true, false, null, null);
+		ChannelInputStream inputStream = TestUtils.getPropertyValue(connection, "channelInputStream", ChannelInputStream.class);
+		inputStream.write(new byte[] {(byte) 0x80}, 1);
+		assertEquals(0x80, inputStream.read());
+	}
+
 }