From a46ac3b8149cfc59f92b6e14b3935cd813acb492 Mon Sep 17 00:00:00 2001 From: baiju Date: Sat, 10 Jun 2017 10:07:23 +0800 Subject: [PATCH] fixbug: soTimeout doesn't work. --- .../client/impl/SimpleCanalConnector.java | 183 ++++++++++-------- 1 file changed, 103 insertions(+), 80 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java index 96cfb618..78daee86 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java +++ b/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 supportedCompressions = new ArrayList(); 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."); } @@ -143,18 +146,19 @@ public class SimpleCanalConnector implements CanalConnector { supportedCompressions.addAll(handshake.getSupportedCompressionsList()); // ClientAuth ca = ClientAuth.newBuilder() - .setUsername(username != null ? username : "") - .setNetReadTimeout(soTimeout) - .setNetWriteTimeout(soTimeout) - .build(); - writeWithHeader(channel, - Packet.newBuilder() - .setType(PacketType.CLIENTAUTHENTICATION) - .setBody(ca.toByteString()) - .build() - .toByteArray()); + .setUsername(username != null ? username : "") + .setPassword(ByteString.copyFromUtf8(password != null ? password : "")) + .setNetReadTimeout(soTimeout) + .setNetWriteTimeout(soTimeout) + .build(); + 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,16 +177,28 @@ 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) { - try { - channel.close(); - } catch (IOException e) { - logger.warn("exception on closing channel:{} \n {}", channel, e); - } + quietlyClose(channel); channel = null; } } + private void quietlyClose(Channel channel) { + try { + channel.close(); + } catch (IOException e) { + logger.warn("exception on closing channel:{} \n {}", channel, e); + } + } + public void subscribe() throws CanalClientException { subscribe(""); // 传递空字符即可 } @@ -190,19 +206,19 @@ public class SimpleCanalConnector implements CanalConnector { public void subscribe(String filter) throws CanalClientException { waitClientRunning(); try { - writeWithHeader(channel, - Packet.newBuilder() - .setType(PacketType.SUBSCRIPTION) - .setBody(Sub.newBuilder() - .setDestination(clientIdentity.getDestination()) - .setClientId(String.valueOf(clientIdentity.getClientId())) - .setFilter(filter != null ? filter : "") - .build() - .toByteString()) - .build() - .toByteArray()); + writeWithHeader( + Packet.newBuilder() + .setType(PacketType.SUBSCRIPTION) + .setBody(Sub.newBuilder() + .setDestination(clientIdentity.getDestination()) + .setClientId(String.valueOf(clientIdentity.getClientId())) + .setFilter(filter != null ? filter : "") + .build() + .toByteString()) + .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,18 +233,18 @@ public class SimpleCanalConnector implements CanalConnector { public void unsubscribe() throws CanalClientException { waitClientRunning(); try { - writeWithHeader(channel, - Packet.newBuilder() - .setType(PacketType.UNSUBSCRIPTION) - .setBody(Unsub.newBuilder() - .setDestination(clientIdentity.getDestination()) - .setClientId(String.valueOf(clientIdentity.getClientId())) - .build() - .toByteString()) - .build() - .toByteArray()); + writeWithHeader( + Packet.newBuilder() + .setType(PacketType.UNSUBSCRIPTION) + .setBody(Unsub.newBuilder() + .setDestination(clientIdentity.getDestination()) + .setClientId(String.valueOf(clientIdentity.getClientId())) + .build() + .toByteString()) + .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,29 +277,28 @@ public class SimpleCanalConnector implements CanalConnector { unit = TimeUnit.MILLISECONDS; } - writeWithHeader(channel, - Packet.newBuilder() - .setType(PacketType.GET) - .setBody(Get.newBuilder() - .setAutoAck(false) - .setDestination(clientIdentity.getDestination()) - .setClientId(String.valueOf(clientIdentity.getClientId())) - .setFetchSize(size) - .setTimeout(time) - .setUnit(unit.ordinal()) - .build() - .toByteString()) - .build() - .toByteArray()); - + writeWithHeader( + Packet.newBuilder() + .setType(PacketType.GET) + .setBody(Get.newBuilder() + .setAutoAck(false) + .setDestination(clientIdentity.getDestination()) + .setClientId(String.valueOf(clientIdentity.getClientId())) + .setFetchSize(size) + .setTimeout(time) + .setUnit(unit.ordinal()) + .build() + .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)) { @@ -310,16 +325,16 @@ public class SimpleCanalConnector implements CanalConnector { public void ack(long batchId) throws CanalClientException { waitClientRunning(); ClientAck ca = ClientAck.newBuilder() - .setDestination(clientIdentity.getDestination()) - .setClientId(String.valueOf(clientIdentity.getClientId())) - .setBatchId(batchId) - .build(); + .setDestination(clientIdentity.getDestination()) + .setClientId(String.valueOf(clientIdentity.getClientId())) + .setBatchId(batchId) + .build(); try { - writeWithHeader(channel, Packet.newBuilder() - .setType(PacketType.CLIENTACK) - .setBody(ca.toByteString()) - .build() - .toByteArray()); + writeWithHeader(Packet.newBuilder() + .setType(PacketType.CLIENTACK) + .setBody(ca.toByteString()) + .build() + .toByteArray()); } catch (IOException e) { throw new CanalClientException(e); } @@ -328,16 +343,16 @@ public class SimpleCanalConnector implements CanalConnector { public void rollback(long batchId) throws CanalClientException { waitClientRunning(); ClientRollback ca = ClientRollback.newBuilder() - .setDestination(clientIdentity.getDestination()) - .setClientId(String.valueOf(clientIdentity.getClientId())) - .setBatchId(batchId) - .build(); + .setDestination(clientIdentity.getDestination()) + .setClientId(String.valueOf(clientIdentity.getClientId())) + .setBatchId(batchId) + .build(); try { - writeWithHeader(channel, Packet.newBuilder() - .setType(PacketType.CLIENTROLLBACK) - .setBody(ca.toByteString()) - .build() - .toByteArray()); + writeWithHeader(Packet.newBuilder() + .setType(PacketType.CLIENTROLLBACK) + .setBody(ca.toByteString()) + .build() + .toByteArray()); } catch (IOException e) { throw new CanalClientException(e); } @@ -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) {