fixbug: soTimeout doesn't work.

client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java

@@ -5,7 +5,7 @@ import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
-import java.nio.channels.SocketChannel;
+import java.nio.channels.*;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -37,7 +37,6 @@ import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.protocol.exception.CanalClientException;
 import com.google.protobuf.ByteString;
-import com.google.protobuf.InvalidProtocolBufferException;
 
 /**
  * 基于{@linkplain CanalServerWithNetty}定义的网络协议接口，对于canal数据进行get/rollback/ack等操作
@@ -57,6 +56,8 @@ public class SimpleCanalConnector implements CanalConnector {
     private final ByteBuffer     readHeader            = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
     private final ByteBuffer     writeHeader           = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
     private SocketChannel        channel;
+    private ReadableByteChannel  readableChannel;
+    private WritableByteChannel  writableChannel;
     private List<Compression>    supportedCompressions = new ArrayList<Compression>();
     private ClientIdentity       clientIdentity;
     private ClientRunningMonitor runningMonitor;                                                             // 运行控制
@@ -130,7 +131,9 @@ public class SimpleCanalConnector implements CanalConnector {
                 address = getNextAddress();
             }
             channel.connect(address);
-            Packet p = Packet.parseFrom(readNextPacket(channel));
+            readableChannel = Channels.newChannel(channel.socket().getInputStream());
+            writableChannel = Channels.newChannel(channel.socket().getOutputStream());
+            Packet p = Packet.parseFrom(readNextPacket());
             if (p.getVersion() != 1) {
                 throw new CanalClientException("unsupported version at this client.");
             }
@@ -144,17 +147,18 @@ public class SimpleCanalConnector implements CanalConnector {
             //
             ClientAuth ca = ClientAuth.newBuilder()
                     .setUsername(username != null ? username : "")
+                    .setPassword(ByteString.copyFromUtf8(password != null ? password : ""))
                     .setNetReadTimeout(soTimeout)
                     .setNetWriteTimeout(soTimeout)
                     .build();
-            writeWithHeader(channel,
+            writeWithHeader(
                     Packet.newBuilder()
                             .setType(PacketType.CLIENTAUTHENTICATION)
                             .setBody(ca.toByteString())
                             .build()
                             .toByteArray());
             //
-            Packet ack = Packet.parseFrom(readNextPacket(channel));
+            Packet ack = Packet.parseFrom(readNextPacket());
             if (ack.getType() != PacketType.ACK) {
                 throw new CanalClientException("unexpected packet type when ack is expected");
             }
@@ -173,14 +177,26 @@ public class SimpleCanalConnector implements CanalConnector {
     }
 
     private void doDisconnnect() throws CanalClientException {
+        if (readableChannel != null) {
+            quietlyClose(readableChannel);
+            readableChannel = null;
+        }
+        if (writableChannel != null) {
+            quietlyClose(writableChannel);
+            writableChannel = null;
+        }
         if (channel != null) {
+            quietlyClose(channel);
+            channel = null;
+        }
+    }
+
+    private void quietlyClose(Channel channel) {
         try {
             channel.close();
         } catch (IOException e) {
             logger.warn("exception on closing channel:{} \n {}", channel, e);
         }
-            channel = null;
-        }
     }
 
     public void subscribe() throws CanalClientException {
@@ -190,7 +206,7 @@ public class SimpleCanalConnector implements CanalConnector {
     public void subscribe(String filter) throws CanalClientException {
         waitClientRunning();
         try {
-            writeWithHeader(channel,
+            writeWithHeader(
                     Packet.newBuilder()
                             .setType(PacketType.SUBSCRIPTION)
                             .setBody(Sub.newBuilder()
@@ -202,7 +218,7 @@ public class SimpleCanalConnector implements CanalConnector {
                             .build()
                             .toByteArray());
             //
-            Packet p = Packet.parseFrom(readNextPacket(channel));
+            Packet p = Packet.parseFrom(readNextPacket());
             Ack ack = Ack.parseFrom(p.getBody());
             if (ack.getErrorCode() > 0) {
                 throw new CanalClientException("failed to subscribe with reason: " + ack.getErrorMessage());
@@ -217,7 +233,7 @@ public class SimpleCanalConnector implements CanalConnector {
     public void unsubscribe() throws CanalClientException {
         waitClientRunning();
         try {
-            writeWithHeader(channel,
+            writeWithHeader(
                     Packet.newBuilder()
                             .setType(PacketType.UNSUBSCRIPTION)
                             .setBody(Unsub.newBuilder()
@@ -228,7 +244,7 @@ public class SimpleCanalConnector implements CanalConnector {
                             .build()
                             .toByteArray());
             //
-            Packet p = Packet.parseFrom(readNextPacket(channel));
+            Packet p = Packet.parseFrom(readNextPacket());
             Ack ack = Ack.parseFrom(p.getBody());
             if (ack.getErrorCode() > 0) {
                 throw new CanalClientException("failed to unSubscribe with reason: " + ack.getErrorMessage());
@@ -261,7 +277,7 @@ public class SimpleCanalConnector implements CanalConnector {
                 unit = TimeUnit.MILLISECONDS;
             }
 
-            writeWithHeader(channel,
+            writeWithHeader(
                     Packet.newBuilder()
                             .setType(PacketType.GET)
                             .setBody(Get.newBuilder()
@@ -275,15 +291,14 @@ public class SimpleCanalConnector implements CanalConnector {
                                     .toByteString())
                             .build()
                             .toByteArray());
-
             return receiveMessages();
         } catch (IOException e) {
             throw new CanalClientException(e);
         }
     }
 
-    private Message receiveMessages() throws InvalidProtocolBufferException, IOException {
-        Packet p = Packet.parseFrom(readNextPacket(channel));
+    private Message receiveMessages() throws IOException {
+        Packet p = Packet.parseFrom(readNextPacket());
         switch (p.getType()) {
             case MESSAGES: {
                 if (!p.getCompression().equals(Compression.NONE)) {
@@ -315,7 +330,7 @@ public class SimpleCanalConnector implements CanalConnector {
                 .setBatchId(batchId)
                 .build();
         try {
-            writeWithHeader(channel, Packet.newBuilder()
+            writeWithHeader(Packet.newBuilder()
                     .setType(PacketType.CLIENTACK)
                     .setBody(ca.toByteString())
                     .build()
@@ -333,7 +348,7 @@ public class SimpleCanalConnector implements CanalConnector {
                 .setBatchId(batchId)
                 .build();
         try {
-            writeWithHeader(channel, Packet.newBuilder()
+            writeWithHeader(Packet.newBuilder()
                     .setType(PacketType.CLIENTROLLBACK)
                     .setBody(ca.toByteString())
                     .build()
@@ -350,7 +365,15 @@ public class SimpleCanalConnector implements CanalConnector {
 
     // ==================== helper method ====================
 
-    private void writeWithHeader(SocketChannel channel, byte[] body) throws IOException {
+    private void writeWithHeader(byte[] body) throws IOException {
+        writeWithHeader(writableChannel, body);
+    }
+
+    private byte[] readNextPacket() throws IOException {
+        return readNextPacket(readableChannel);
+    }
+
+    private void writeWithHeader(WritableByteChannel channel, byte[] body) throws IOException {
         synchronized (writeDataLock) {
             writeHeader.clear();
             writeHeader.putInt(body.length);
@@ -360,7 +383,7 @@ public class SimpleCanalConnector implements CanalConnector {
         }
     }
 
-    private byte[] readNextPacket(SocketChannel channel) throws IOException {
+    private byte[] readNextPacket(ReadableByteChannel channel) throws IOException {
         synchronized (readDataLock) {
             readHeader.clear();
             read(channel, readHeader);
@@ -371,7 +394,7 @@ public class SimpleCanalConnector implements CanalConnector {
         }
     }
 
-    private void read(SocketChannel channel, ByteBuffer buffer) throws IOException {
+    private void read(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
         while (buffer.hasRemaining()) {
             int r = channel.read(buffer);
             if (r == -1) {