Error handling and shutdown improvements

README.md

@@ -279,6 +279,40 @@ client.closeUserDataStream(listenKey);
 BinanceApiWebSocketClient client = BinanceApiClientFactory.newInstance().newWebSocketClient();
 ```
 
+#### Handling web socket errors
+
+Each of the methods on `BinanceApiWebSocketClient`, which opens a new web socket, takes a `BinanceApiCallback`, which is
+called for each event received from the Binance servers. 
+
+The `BinanceApiCallback` interface also has a `onFailure(Throwable)` method, which, optionally, can be implemented to 
+receive notifications if the web-socket fails, e.g. disconnection.   
+
+```java
+client.onAggTradeEvent(symbol.toLowerCase(), new BinanceApiCallback<AggTradeEvent>() {
+    @Override
+    public void onResponse(final AggTradeEvent response) {
+        System.out.println(response);
+    }
+
+    @Override
+    public void onFailure(final Throwable cause) {
+        System.err.println("Web socket failed");
+        cause.printStackTrace(System.err);
+    }
+});
+```
+
+#### Closing web sockets
+
+Each of the methods on `BinanceApiWebSocketClient`, which opens a new web socket, also returns a `Closeable`.
+This `Closeable` can be used to close the underlying web socket and free any associated resources, e.g.
+
+```java
+Closable ws = client.onAggTradeEvent("ethbtc", someCallback);
+// some time later...
+ws.close();
+```
+
 #### Listen for aggregated trade events for ETH/BTC
 ```java
 client.onAggTradeEvent("ethbtc", (AggTradeEvent response) -> {

src/main/java/com/binance/api/client/BinanceApiCallback.java

@@ -1,19 +1,24 @@
 package com.binance.api.client;
 
-import com.binance.api.client.exception.BinanceApiException;
-
 /**
  * BinanceApiCallback is a functional interface used together with the BinanceApiAsyncClient to provide a non-blocking REST client.
  *
  * @param <T> the return type from the callback
  */
+@FunctionalInterface
 public interface BinanceApiCallback<T> {
 
     /**
      * Called whenever a response comes back from the Binance API.
      *
      * @param response the expected response object
-     * @throws BinanceApiException if it is not possible to obtain the expected response object (e.g. incorrect API-KEY).
      */
-    void onResponse(T response) throws BinanceApiException;
+    void onResponse(T response);
+
+    /**
+     * Called whenever an error occurs.
+     *
+     * @param cause the cause of the failure
+     */
+    default void onFailure(Throwable cause) {}
 }

src/main/java/com/binance/api/client/BinanceApiWebSocketClient.java

@@ -1,25 +1,64 @@
 package com.binance.api.client;
 
-import java.util.List;
 import com.binance.api.client.domain.event.AggTradeEvent;
 import com.binance.api.client.domain.event.AllMarketTickersEvent;
 import com.binance.api.client.domain.event.CandlestickEvent;
 import com.binance.api.client.domain.event.DepthEvent;
 import com.binance.api.client.domain.event.UserDataUpdateEvent;
 import com.binance.api.client.domain.market.CandlestickInterval;
 
+import java.io.Closeable;
+import java.util.List;
+
 /**
  * Binance API data streaming façade, supporting streaming of events through web sockets.
  */
-public interface BinanceApiWebSocketClient {
+public interface BinanceApiWebSocketClient extends Closeable {
+
+    /**
+     * Open a new web socket to receive {@link DepthEvent depthEvents} on a callback.
+     *
+     * @param symbol   the market symbol to subscribe to
+     * @param callback the callback to call on new events
+     * @return a {@link Closeable} that allows the underlying web socket to be closed.
+     */
+    Closeable onDepthEvent(String symbol, BinanceApiCallback<DepthEvent> callback);
 
-  void onDepthEvent(String symbol, BinanceApiCallback<DepthEvent> callback);
+    /**
+     * Open a new web socket to receive {@link CandlestickEvent candlestickEvents} on a callback.
+     *
+     * @param symbol   the market symbol to subscribe to
+     * @param interval the interval of the candles tick events required
+     * @param callback the callback to call on new events
+     * @return a {@link Closeable} that allows the underlying web socket to be closed.
+     */
+    Closeable onCandlestickEvent(String symbol, CandlestickInterval interval, BinanceApiCallback<CandlestickEvent> callback);
 
-  void onCandlestickEvent(String symbol, CandlestickInterval interval, BinanceApiCallback<CandlestickEvent> callback);
+    /**
+     * Open a new web socket to receive {@link AggTradeEvent aggTradeEvents} on a callback.
+     *
+     * @param symbol   the market symbol to subscribe to
+     * @param callback the callback to call on new events
+     * @return a {@link Closeable} that allows the underlying web socket to be closed.
+     */
+    Closeable onAggTradeEvent(String symbol, BinanceApiCallback<AggTradeEvent> callback);
 
-  void onAggTradeEvent(String symbol, BinanceApiCallback<AggTradeEvent> callback);
+    /**
+     * Open a new web socket to receive {@link UserDataUpdateEvent userDataUpdateEvents} on a callback.
+     *
+     * @param listenKey the listen key to subscribe to.
+     * @param callback  the callback to call on new events
+     * @return a {@link Closeable} that allows the underlying web socket to be closed.
+     */
+    Closeable onUserDataUpdateEvent(String listenKey, BinanceApiCallback<UserDataUpdateEvent> callback);
 
-  void onUserDataUpdateEvent(String listenKey, BinanceApiCallback<UserDataUpdateEvent> callback);
+    /**
+     * Open a new web socket to receive {@link AllMarketTickersEvent allMarketTickersEvents} on a callback.
+     *
+     * @param callback the callback to call on new events
+     * @return a {@link Closeable} that allows the underlying web socket to be closed.
+     */
+    Closeable onAllMarketTickersEvent(BinanceApiCallback<List<AllMarketTickersEvent>> callback);
 
-  void onAllMarketTickersEvent(BinanceApiCallback<List<AllMarketTickersEvent>> callback);
+    void close();
 }

src/main/java/com/binance/api/client/impl/BinanceApiWebSocketClientImpl.java

@@ -1,11 +1,5 @@
 package com.binance.api.client.impl;
 
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.List;
-import okhttp3.Dispatcher;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
 import com.binance.api.client.BinanceApiCallback;
 import com.binance.api.client.BinanceApiWebSocketClient;
 import com.binance.api.client.constant.BinanceApiConstants;
@@ -15,54 +9,66 @@
 import com.binance.api.client.domain.event.DepthEvent;
 import com.binance.api.client.domain.event.UserDataUpdateEvent;
 import com.binance.api.client.domain.market.CandlestickInterval;
+import okhttp3.Dispatcher;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.WebSocket;
+
+import java.io.Closeable;
+import java.util.List;
 
 /**
  * Binance API WebSocket client implementation using OkHttp.
  */
 public class BinanceApiWebSocketClientImpl implements BinanceApiWebSocketClient, Closeable {
 
-  private OkHttpClient client;
-
-  public BinanceApiWebSocketClientImpl() {
-    Dispatcher d = new Dispatcher();
-    d.setMaxRequestsPerHost(100);
-    this.client = new OkHttpClient.Builder().dispatcher(d).build();
-  }
+    private OkHttpClient client;
 
-  public void onDepthEvent(String symbol, BinanceApiCallback<DepthEvent> callback) {
-    final String channel = String.format("%s@depth", symbol);
-    createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, DepthEvent.class));
-  }
+    public BinanceApiWebSocketClientImpl() {
+        Dispatcher d = new Dispatcher();
+        d.setMaxRequestsPerHost(100);
+        this.client = new OkHttpClient.Builder().dispatcher(d).build();
+    }
 
-  @Override
-  public void onCandlestickEvent(String symbol, CandlestickInterval interval, BinanceApiCallback<CandlestickEvent> callback) {
-    final String channel = String.format("%s@kline_%s", symbol, interval.getIntervalId());
-    createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, CandlestickEvent.class));
-  }
+    public Closeable onDepthEvent(String symbol, BinanceApiCallback<DepthEvent> callback) {
+        final String channel = String.format("%s@depth", symbol);
+        return createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, DepthEvent.class));
+    }
 
-  public void onAggTradeEvent(String symbol, BinanceApiCallback<AggTradeEvent> callback) {
-    final String channel = String.format("%s@aggTrade", symbol);
-    createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, AggTradeEvent.class));
-  }
+    @Override
+    public Closeable onCandlestickEvent(String symbol, CandlestickInterval interval, BinanceApiCallback<CandlestickEvent> callback) {
+        final String channel = String.format("%s@kline_%s", symbol, interval.getIntervalId());
+        return createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, CandlestickEvent.class));
+    }
 
-  public void onUserDataUpdateEvent(String listenKey, BinanceApiCallback<UserDataUpdateEvent> callback) {
-    createNewWebSocket(listenKey, new BinanceApiWebSocketListener<>(callback, UserDataUpdateEvent.class));
-  }
+    public Closeable onAggTradeEvent(String symbol, BinanceApiCallback<AggTradeEvent> callback) {
+        final String channel = String.format("%s@aggTrade", symbol);
+        return createNewWebSocket(channel, new BinanceApiWebSocketListener<>(callback, AggTradeEvent.class));
+    }
 
-  public void onAllMarketTickersEvent(BinanceApiCallback<List<AllMarketTickersEvent>> callback) {
-    final String channel = "!ticker@arr";
-    createNewWebSocket(channel, new BinanceApiWebSocketListener<List<AllMarketTickersEvent>>(callback));
-  }
+    public Closeable onUserDataUpdateEvent(String listenKey, BinanceApiCallback<UserDataUpdateEvent> callback) {
+        return createNewWebSocket(listenKey, new BinanceApiWebSocketListener<>(callback, UserDataUpdateEvent.class));
+    }
 
-  private void createNewWebSocket(String channel, BinanceApiWebSocketListener<?> listener) {
-    String streamingUrl = String.format("%s/%s", BinanceApiConstants.WS_API_BASE_URL, channel);
-    Request request = new Request.Builder().url(streamingUrl).build();
-    client.newWebSocket(request, listener);
-  }
+    public Closeable onAllMarketTickersEvent(BinanceApiCallback<List<AllMarketTickersEvent>> callback) {
+        final String channel = "!ticker@arr";
+        return createNewWebSocket(channel, new BinanceApiWebSocketListener<List<AllMarketTickersEvent>>(callback));
+    }
 
-  @Override
-  public void close() throws IOException {
-    client.dispatcher().executorService().shutdown();
-  }
+    @Override
+    public void close() {
+        client.dispatcher().executorService().shutdown();
+    }
 
+    private Closeable createNewWebSocket(String channel, BinanceApiWebSocketListener<?> listener) {
+        String streamingUrl = String.format("%s/%s", BinanceApiConstants.WS_API_BASE_URL, channel);
+        Request request = new Request.Builder().url(streamingUrl).build();
+        final WebSocket webSocket = client.newWebSocket(request, listener);
+        return () -> {
+            final int code = 1000;
+            listener.onClosing(webSocket, code, null);
+            webSocket.close(code, null);
+            listener.onClosed(webSocket, code, null);
+        };
+    }
 }

src/main/java/com/binance/api/client/impl/BinanceApiWebSocketListener.java

@@ -20,6 +20,8 @@ public class BinanceApiWebSocketListener<T> extends WebSocketListener {
 
   private TypeReference<T> eventTypeReference;
 
+  private boolean closing = false;
+
   public BinanceApiWebSocketListener(BinanceApiCallback<T> callback, Class<T> eventClass) {
     this.callback = callback;
     this.eventClass = eventClass;
@@ -46,8 +48,15 @@ public void onMessage(WebSocket webSocket, String text) {
     }
   }
 
+  @Override
+  public void onClosing(final WebSocket webSocket, final int code, final String reason) {
+    closing = true;
+  }
+
   @Override
   public void onFailure(WebSocket webSocket, Throwable t, Response response) {
-    throw new BinanceApiException(t);
+    if (!closing) {
+      callback.onFailure(t);
+    }
   }
 }