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

Merge branch 'master' of https://github.com/rewerma/canal

Browse Source
This commit is contained in:
mcy
2018-09-14 11:24:27 +08:00
parent e8e0032404 abf6a37f3a
commit f3d1c8b754
9 changed files with 15943 additions and 14201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -159,7 +159,7 @@ public class SimpleCanalConnector implements CanalConnector {
}
//
Handshake handshake = Handshake.parseFrom(p.getBody());
supportedCompressions.addAll(handshake.getSupportedCompressionsList());
supportedCompressions.add(handshake.getSupportedCompressions());
//
ClientAuth ca = ClientAuth.newBuilder()
.setUsername(username != null ? username : "")

188
client/src/test/java/com/alibaba/otter/canal/client/running/ClientRunningTest.java
View File

@@ -20,117 +20,117 @@ import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
public class ClientRunningTest extends AbstractZkTest {
private ZkClientx zkclientx = new ZkClientx(cluster1 + ";" + cluster2);
private short clientId = 1001;
private ZkClientx zkclientx = new ZkClientx(cluster1 + ";" + cluster2);
private short clientId = 1001;
@Before
public void setUp() {
String path = ZookeeperPathUtils.getDestinationPath(destination);
zkclientx.deleteRecursive(path);
@Before
public void setUp() {
String path = ZookeeperPathUtils.getDestinationPath(destination);
zkclientx.deleteRecursive(path);
zkclientx.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientId), true);
}
zkclientx.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientId), true);
}
@After
public void tearDown() {
String path = ZookeeperPathUtils.getDestinationPath(destination);
zkclientx.deleteRecursive(path);
}
@After
public void tearDown() {
String path = ZookeeperPathUtils.getDestinationPath(destination);
zkclientx.deleteRecursive(path);
}
@Test
public void testOneServer() {
final CountDownLatch countLatch = new CountDownLatch(2);
ClientRunningMonitor runningMonitor = buildClientRunning(countLatch, clientId, 2088);
runningMonitor.start();
sleep(2000L);
runningMonitor.stop();
sleep(2000L);
@Test
public void testOneServer() {
final CountDownLatch countLatch = new CountDownLatch(2);
ClientRunningMonitor runningMonitor = buildClientRunning(countLatch, clientId, 2088);
runningMonitor.start();
sleep(2000L);
runningMonitor.stop();
sleep(2000L);
if (countLatch.getCount() != 0) {
Assert.fail();
}
}
if (countLatch.getCount() != 0) {
Assert.fail();
}
}
@Test
public void testMultiServer() {
final CountDownLatch countLatch = new CountDownLatch(30);
final ClientRunningMonitor runningMonitor1 = buildClientRunning(countLatch, clientId, 2088);
final ClientRunningMonitor runningMonitor2 = buildClientRunning(countLatch, clientId, 2089);
final ClientRunningMonitor runningMonitor3 = buildClientRunning(countLatch, clientId, 2090);
final ExecutorService executor = Executors.newFixedThreadPool(3);
executor.submit(new Runnable() {
@Test
public void testMultiServer() {
final CountDownLatch countLatch = new CountDownLatch(30);
final ClientRunningMonitor runningMonitor1 = buildClientRunning(countLatch, clientId, 2088);
final ClientRunningMonitor runningMonitor2 = buildClientRunning(countLatch, clientId, 2089);
final ClientRunningMonitor runningMonitor3 = buildClientRunning(countLatch, clientId, 2090);
final ExecutorService executor = Executors.newFixedThreadPool(3);
executor.submit(new Runnable() {
public void run() {
for (int i = 0; i < 10; i++) {
if (!runningMonitor1.isStart()) {
runningMonitor1.start();
}
sleep(2000L + RandomUtils.nextInt(500));
runningMonitor1.stop();
sleep(2000L + RandomUtils.nextInt(500));
}
}
public void run() {
for (int i = 0; i < 10; i++) {
if (!runningMonitor1.isStart()) {
runningMonitor1.start();
}
sleep(2000L + RandomUtils.nextInt(500));
runningMonitor1.stop();
sleep(2000L + RandomUtils.nextInt(500));
}
}
});
});
executor.submit(new Runnable() {
executor.submit(new Runnable() {
public void run() {
for (int i = 0; i < 10; i++) {
if (!runningMonitor2.isStart()) {
runningMonitor2.start();
}
sleep(2000L + RandomUtils.nextInt(500));
runningMonitor2.stop();
sleep(2000L + RandomUtils.nextInt(500));
}
}
public void run() {
for (int i = 0; i < 10; i++) {
if (!runningMonitor2.isStart()) {
runningMonitor2.start();
}
sleep(2000L + RandomUtils.nextInt(500));
runningMonitor2.stop();
sleep(2000L + RandomUtils.nextInt(500));
}
}
});
});
executor.submit(new Runnable() {
executor.submit(new Runnable() {
public void run() {
for (int i = 0; i < 10; i++) {
if (!runningMonitor3.isStart()) {
runningMonitor3.start();
}
sleep(2000L + RandomUtils.nextInt(500));
runningMonitor3.stop();
sleep(2000L + RandomUtils.nextInt(500));
}
}
public void run() {
for (int i = 0; i < 10; i++) {
if (!runningMonitor3.isStart()) {
runningMonitor3.start();
}
sleep(2000L + RandomUtils.nextInt(500));
runningMonitor3.stop();
sleep(2000L + RandomUtils.nextInt(500));
}
}
});
});
sleep(30000L);
}
sleep(30000L);
}
private ClientRunningMonitor buildClientRunning(final CountDownLatch countLatch, final short clientId,
final int port) {
ClientRunningData clientData = new ClientRunningData();
clientData.setClientId(clientId);
clientData.setAddress(AddressUtils.getHostIp());
private ClientRunningMonitor buildClientRunning(final CountDownLatch countLatch, final short clientId,
final int port) {
ClientRunningData clientData = new ClientRunningData();
clientData.setClientId(clientId);
clientData.setAddress(AddressUtils.getHostIp());
ClientRunningMonitor runningMonitor = new ClientRunningMonitor();
runningMonitor.setDestination(destination);
runningMonitor.setZkClient(zkclientx);
runningMonitor.setClientData(clientData);
runningMonitor.setListener(new ClientRunningListener() {
ClientRunningMonitor runningMonitor = new ClientRunningMonitor();
runningMonitor.setDestination(destination);
runningMonitor.setZkClient(zkclientx);
runningMonitor.setClientData(clientData);
runningMonitor.setListener(new ClientRunningListener() {
public InetSocketAddress processActiveEnter() {
System.out.println(String.format("clientId:%s port:%s has start", clientId, port));
countLatch.countDown();
return new InetSocketAddress(AddressUtils.getHostIp(), port);
}
public InetSocketAddress processActiveEnter() {
System.out.println(String.format("clientId:%s port:%s has start", clientId, port));
countLatch.countDown();
return new InetSocketAddress(AddressUtils.getHostIp(), port);
}
public void processActiveExit() {
countLatch.countDown();
System.out.println(String.format("clientId:%s port:%s has stop", clientId, port));
}
public void processActiveExit() {
countLatch.countDown();
System.out.println(String.format("clientId:%s port:%s has stop", clientId, port));
}
});
runningMonitor.setDelayTime(1);
return runningMonitor;
}
}
});
runningMonitor.setDelayTime(1);
return runningMonitor;
}
}