Merge pull request #623 from rewerma/master

在CanalConnector接口加入stopRunning方法，用于优雅停止client

client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java

@@ -150,4 +150,9 @@ public interface CanalConnector {
      */
     void rollback() throws CanalClientException;
 
+    /**
+     * 中断的阻塞，用于优雅停止client
+     * @throws CanalClientException
+     */
+    void stopRunning() throws CanalClientException;
 }

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

@@ -328,4 +328,8 @@ public class ClusterCanalConnector implements CanalConnector {
         return currentConnector;
     }
 
+    public void stopRunning() {
+        currentConnector.stopRunning();
+    }
+
 }

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

@@ -75,6 +75,8 @@ public class SimpleCanalConnector implements CanalConnector {
     private Object               readDataLock          = new Object();
     private Object               writeDataLock         = new Object();
 
+    private boolean              running               = false;
+
     public SimpleCanalConnector(SocketAddress address, String username, String password, String destination){
         this(address, username, password, destination, 60000);
     }
@@ -99,6 +101,9 @@ public class SimpleCanalConnector implements CanalConnector {
             }
         } else {
             waitClientRunning();
+            if (!running) {
+                return;
+            }
             doConnect();
             if (filter != null) { // 如果存在条件，说明是自动切换，基于上一次的条件订阅一次
                 subscribe(filter);
@@ -208,6 +213,9 @@ public class SimpleCanalConnector implements CanalConnector {
 
     public void subscribe(String filter) throws CanalClientException {
         waitClientRunning();
+        if (!running) {
+            return;
+        }
         try {
             writeWithHeader(Packet.newBuilder()
                 .setType(PacketType.SUBSCRIPTION)
@@ -234,6 +242,9 @@ public class SimpleCanalConnector implements CanalConnector {
 
     public void unsubscribe() throws CanalClientException {
         waitClientRunning();
+        if (!running) {
+            return;
+        }
         try {
             writeWithHeader(Packet.newBuilder()
                 .setType(PacketType.UNSUBSCRIPTION)
@@ -445,6 +456,8 @@ public class SimpleCanalConnector implements CanalConnector {
                     throw new CanalClientException("should connect first");
                 }
 
+                running = true;
+
                 mutex.get();// 阻塞等待
             }
         } catch (InterruptedException e) {
@@ -502,4 +515,13 @@ public class SimpleCanalConnector implements CanalConnector {
         this.filter = filter;
     }
 
+    public void stopRunning() {
+        if (running) {
+            running = false;  //设置为非running状态
+            if (!mutex.state()) {
+                mutex.set(true);  //中断阻塞
+            }
+        }
+    }
+
 }

pom.xml

@@ -254,7 +254,7 @@
             <dependency>
                 <groupId>com.alibaba.fastsql</groupId>
                 <artifactId>fastsql</artifactId>
-                <version>2.0.0_preview_228</version>
+                <version>2.0.0_preview_186</version>
             </dependency>
             <dependency>
                 <groupId>com.alibaba</groupId>