Merge pull request #197 from binance/websocket_stop

Add method to common to stop websocket client

Author: theo-s68

clients/common/pom.xml

@@ -10,6 +10,6 @@
 
     <artifactId>binance-common</artifactId>
     <name>common</name>
-    <version>2.1.1</version>
+    <version>2.2.1</version>
     <packaging>jar</packaging>
 </project>

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/ConnectionInterface.java

@@ -24,4 +24,6 @@ public interface ConnectionInterface {
     void setLogonMethods(List<String> logonMethods);
 
     void setLogoutMethods(List<String> logoutMethods);
+
+    void stop() throws Exception;
 }

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/ConnectionWrapper.java

@@ -49,6 +49,7 @@
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.http.HttpHeader;
+import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
 import org.eclipse.jetty.websocket.api.WebSocketListener;
@@ -92,6 +93,8 @@ public class ConnectionWrapper implements WebSocketListener, ConnectionInterface
 
     private List<BlockingQueue<String>> streamQueues = new ArrayList<>();
 
+    private Timer timer;
+
     public ConnectionWrapper(WebSocketClientConfiguration configuration, Gson gson) {
         this(configuration, null, gson);
     }
@@ -160,8 +163,8 @@ public ConnectionWrapper(
         }
 
         Integer reconnectAfter = configuration.getReconnectIntervalTime();
-        new Timer()
-                .scheduleAtFixedRate(
+        this.timer = new Timer();
+        this.timer.scheduleAtFixedRate(
                         new TimerTask() {
                             @Override
                             public void run() {
@@ -221,6 +224,7 @@ public void connect(Consumer<Session> customCallback)
         CompletableFuture<Session> clientSessionPromise =
                 webSocketClient.connect(this, serverURI, clientUpgradeRequest);
         Session session = clientSessionPromise.join();
+
         if (callback != null) {
             callback.accept(session);
         }
@@ -501,6 +505,7 @@ protected void afterConnect(Session session) {
         if (this.oldSession != null) {
             this.oldSession.close(StatusCode.NORMAL, "close after reconnect", WriteCallback.NOOP);
         }
+        canReconnect = true;
         setReady(true);
     }
 
@@ -548,7 +553,17 @@ public URI getUri(String uri) throws URISyntaxException {
     public void disconnect() {
         if (this.session != null) {
             this.session.disconnect();
+        }
+        setReady(false);
+        canReconnect = false;
+    }
+
+    public void stop() throws Exception {
+        if (this.webSocketClient != null) {
             setReady(false);
+            canReconnect = false;
+            timer.cancel();
+            webSocketClient.stop();
         }
     }
 }

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/PoolConnectionWrapper.java

@@ -147,4 +147,12 @@ public void setLogoutMethods(List<String> logoutMethods) {
     public boolean isConnected() {
         return isConnected;
     }
+
+    @Override
+    public void stop() throws Exception {
+        for (ConnectionWrapper connectionWrapper : connectionList) {
+            connectionWrapper.stop();
+        }
+        isConnected = false;
+    }
 }

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/stream/StreamConnectionInterface.java

@@ -21,4 +21,6 @@ Map<String, StreamBlockingQueue<String>> subscribe(
     boolean isConnected();
 
     void setUserAgent(String userAgent);
+
+    void stop() throws Exception;
 }

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/stream/StreamConnectionPoolWrapper.java

@@ -154,4 +154,12 @@ public void setUserAgent(String userAgent) {
     public boolean isConnected() {
         return isConnected;
     }
+
+    @Override
+    public void stop() throws Exception {
+        for (ConnectionWrapper connectionWrapper : connectionList) {
+            connectionWrapper.stop();
+        }
+        isConnected = false;
+    }
 }

clients/common/src/main/java/com/binance/connector/client/common/websocket/adapter/stream/StreamConnectionWrapper.java

@@ -129,7 +129,7 @@ public void onWebSocketText(String message) {
 
     protected void beforeConnect() {
         // no session, so this is not a reconnect
-        if (this.session == null) {
+        if (this.session == null || !this.session.isOpen()) {
             return;
         }
         RequestWrapperDTO<Object, HashSet<String>> listSubscriptions =

clients/derivatives-trading-coin-futures/docs/rest-api/migration-guide.md

@@ -22,7 +22,7 @@ With the transition to a modularized structure, the Binance Connector has been s
 <dependency>
   <groupId>io.github.binance</groupId>
   <artifactId>binance-derivatives-trading-coin-futures</artifactId>
-  <version>4.0.0</version>
+  <version>4.0.1</version>
 </dependency>
 ```
 

clients/derivatives-trading-coin-futures/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <artifactId>binance-derivatives-trading-coin-futures</artifactId>
     <name>derivatives-trading-coin-futures</name>
-    <version>4.0.0</version>
+    <version>4.0.1</version>
     <packaging>jar</packaging>
 
     <parent>
@@ -31,7 +31,7 @@
         <dependency>
             <groupId>io.github.binance</groupId>
             <artifactId>binance-common</artifactId>
-            <version>2.1.1</version>
+            <version>2.2.1</version>
         </dependency>
     </dependencies>
 </project>

clients/derivatives-trading-options/docs/rest-api/migration-guide.md

@@ -22,7 +22,7 @@ With the transition to a modularized structure, the Binance Connector has been s
 <dependency>
   <groupId>io.github.binance</groupId>
   <artifactId>binance-derivatives-trading-options</artifactId>
-  <version>4.0.0</version>
+  <version>4.0.1</version>
 </dependency>
 ```