chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed format

Browse Source
This commit is contained in:
agapple
2017-07-05 12:46:32 +08:00
parent b265bf4669
commit 62e8ff6ec8
1 changed files with 63 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,7 +5,11 @@ import java.net.InetSocketAddress;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.nio.ByteBuffer; import java.nio.ByteBuffer;
import java.nio.ByteOrder; import java.nio.ByteOrder;
import java.nio.channels.*; import java.nio.channels.Channel;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@@ -146,17 +150,16 @@ public class SimpleCanalConnector implements CanalConnector {
supportedCompressions.addAll(handshake.getSupportedCompressionsList()); supportedCompressions.addAll(handshake.getSupportedCompressionsList());
// //
ClientAuth ca = ClientAuth.newBuilder() ClientAuth ca = ClientAuth.newBuilder()
.setUsername(username != null ? username : "") .setUsername(username != null ? username : "")
.setPassword(ByteString.copyFromUtf8(password != null ? password : "")) .setPassword(ByteString.copyFromUtf8(password != null ? password : ""))
.setNetReadTimeout(soTimeout) .setNetReadTimeout(soTimeout)
.setNetWriteTimeout(soTimeout) .setNetWriteTimeout(soTimeout)
.build(); .build();
writeWithHeader( writeWithHeader(Packet.newBuilder()
Packet.newBuilder() .setType(PacketType.CLIENTAUTHENTICATION)
.setType(PacketType.CLIENTAUTHENTICATION) .setBody(ca.toByteString())
.setBody(ca.toByteString()) .build()
.build() .toByteArray());
.toByteArray());
// //
Packet ack = Packet.parseFrom(readNextPacket()); Packet ack = Packet.parseFrom(readNextPacket());
if (ack.getType() != PacketType.ACK) { if (ack.getType() != PacketType.ACK) {
@@ -206,17 +209,16 @@ public class SimpleCanalConnector implements CanalConnector {
public void subscribe(String filter) throws CanalClientException { public void subscribe(String filter) throws CanalClientException {
waitClientRunning(); waitClientRunning();
try { try {
writeWithHeader( writeWithHeader(Packet.newBuilder()
Packet.newBuilder() .setType(PacketType.SUBSCRIPTION)
.setType(PacketType.SUBSCRIPTION) .setBody(Sub.newBuilder()
.setBody(Sub.newBuilder() .setDestination(clientIdentity.getDestination())
.setDestination(clientIdentity.getDestination()) .setClientId(String.valueOf(clientIdentity.getClientId()))
.setClientId(String.valueOf(clientIdentity.getClientId())) .setFilter(filter != null ? filter : "")
.setFilter(filter != null ? filter : "") .build()
.build() .toByteString())
.toByteString()) .build()
.build() .toByteArray());
.toByteArray());
// //
Packet p = Packet.parseFrom(readNextPacket()); Packet p = Packet.parseFrom(readNextPacket());
Ack ack = Ack.parseFrom(p.getBody()); Ack ack = Ack.parseFrom(p.getBody());
@@ -233,16 +235,15 @@ public class SimpleCanalConnector implements CanalConnector {
public void unsubscribe() throws CanalClientException { public void unsubscribe() throws CanalClientException {
waitClientRunning(); waitClientRunning();
try { try {
writeWithHeader( writeWithHeader(Packet.newBuilder()
Packet.newBuilder() .setType(PacketType.UNSUBSCRIPTION)
.setType(PacketType.UNSUBSCRIPTION) .setBody(Unsub.newBuilder()
.setBody(Unsub.newBuilder() .setDestination(clientIdentity.getDestination())
.setDestination(clientIdentity.getDestination()) .setClientId(String.valueOf(clientIdentity.getClientId()))
.setClientId(String.valueOf(clientIdentity.getClientId())) .build()
.build() .toByteString())
.toByteString()) .build()
.build() .toByteArray());
.toByteArray());
// //
Packet p = Packet.parseFrom(readNextPacket()); Packet p = Packet.parseFrom(readNextPacket());
Ack ack = Ack.parseFrom(p.getBody()); Ack ack = Ack.parseFrom(p.getBody());
@@ -277,20 +278,19 @@ public class SimpleCanalConnector implements CanalConnector {
unit = TimeUnit.MILLISECONDS; unit = TimeUnit.MILLISECONDS;
} }
writeWithHeader( writeWithHeader(Packet.newBuilder()
Packet.newBuilder() .setType(PacketType.GET)
.setType(PacketType.GET) .setBody(Get.newBuilder()
.setBody(Get.newBuilder() .setAutoAck(false)
.setAutoAck(false) .setDestination(clientIdentity.getDestination())
.setDestination(clientIdentity.getDestination()) .setClientId(String.valueOf(clientIdentity.getClientId()))
.setClientId(String.valueOf(clientIdentity.getClientId())) .setFetchSize(size)
.setFetchSize(size) .setTimeout(time)
.setTimeout(time) .setUnit(unit.ordinal())
.setUnit(unit.ordinal()) .build()
.build() .toByteString())
.toByteString()) .build()
.build() .toByteArray());
.toByteArray());
return receiveMessages(); return receiveMessages();
} catch (IOException e) { } catch (IOException e) {
throw new CanalClientException(e); throw new CanalClientException(e);
@@ -325,16 +325,16 @@ public class SimpleCanalConnector implements CanalConnector {
public void ack(long batchId) throws CanalClientException { public void ack(long batchId) throws CanalClientException {
waitClientRunning(); waitClientRunning();
ClientAck ca = ClientAck.newBuilder() ClientAck ca = ClientAck.newBuilder()
.setDestination(clientIdentity.getDestination()) .setDestination(clientIdentity.getDestination())
.setClientId(String.valueOf(clientIdentity.getClientId())) .setClientId(String.valueOf(clientIdentity.getClientId()))
.setBatchId(batchId) .setBatchId(batchId)
.build(); .build();
try { try {
writeWithHeader(Packet.newBuilder() writeWithHeader(Packet.newBuilder()
.setType(PacketType.CLIENTACK) .setType(PacketType.CLIENTACK)
.setBody(ca.toByteString()) .setBody(ca.toByteString())
.build() .build()
.toByteArray()); .toByteArray());
} catch (IOException e) { } catch (IOException e) {
throw new CanalClientException(e); throw new CanalClientException(e);
} }
@@ -343,16 +343,16 @@ public class SimpleCanalConnector implements CanalConnector {
public void rollback(long batchId) throws CanalClientException { public void rollback(long batchId) throws CanalClientException {
waitClientRunning(); waitClientRunning();
ClientRollback ca = ClientRollback.newBuilder() ClientRollback ca = ClientRollback.newBuilder()
.setDestination(clientIdentity.getDestination()) .setDestination(clientIdentity.getDestination())
.setClientId(String.valueOf(clientIdentity.getClientId())) .setClientId(String.valueOf(clientIdentity.getClientId()))
.setBatchId(batchId) .setBatchId(batchId)
.build(); .build();
try { try {
writeWithHeader(Packet.newBuilder() writeWithHeader(Packet.newBuilder()
.setType(PacketType.CLIENTROLLBACK) .setType(PacketType.CLIENTROLLBACK)
.setBody(ca.toByteString()) .setBody(ca.toByteString())
.build() .build()
.toByteArray()); .toByteArray());
} catch (IOException e) { } catch (IOException e) {
throw new CanalClientException(e); throw new CanalClientException(e);
} }