From 4141049bb90ce6dc1ce05a58c95f2d974fdd5865 Mon Sep 17 00:00:00 2001
From: rewerma <rewerma@163.com>
Date: Mon, 7 May 2018 15:13:24 +0800
Subject: [PATCH] =?UTF-8?q?=E5=9C=A8CanalConnector=E6=8E=A5=E5=8F=A3?=
 =?UTF-8?q?=E5=8A=A0=E5=85=A5stopRunning=E6=96=B9=E6=B3=95=EF=BC=8C?=
 =?UTF-8?q?=E7=94=A8=E4=BA=8E=E4=BC=98=E9=9B=85=E5=81=9C=E6=AD=A2client?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../otter/canal/client/CanalConnector.java    |  5 ++++
 .../client/impl/ClusterCanalConnector.java    |  4 ++++
 .../client/impl/SimpleCanalConnector.java     | 24 ++++++++++++++++++-
 pom.xml                                       |  2 +-
 4 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java
index fb0db67f..85177d98 100644
--- a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java
+++ b/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;
 }
diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java
index c339af78..8446e948 100644
--- a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java
+++ b/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();
+    }
+
 }
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 3b25e5d2..a6eabed1 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
@@ -44,7 +44,7 @@ import com.google.protobuf.ByteString;
 
 /**
  * 基于{@linkplain CanalServerWithNetty}定义的网络协议接口，对于canal数据进行get/rollback/ack等操作
- * 
+ *
  * @author jianghang 2012-10-24 下午05:37:20
  * @version 1.0.0
  */
@@ -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);  //中断阻塞
+            }
+        }
+    }
+
 }
diff --git a/pom.xml b/pom.xml
index 8c93c9ee..e9c2623b 100644
--- a/pom.xml
+++ b/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>