package info.bitrich.xchangestream.binance;

import com.fasterxml.jackson.databind.JsonNode;

import info.bitrich.xchangestream.binance.dto.BaseBinanceWebSocketTransaction.BinanceWebSocketTypes;
import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;

import io.reactivex.Observable;

import java.io.IOException;

public class BinanceUserDataStreamingService extends JsonNettyStreamingService {

	private static final String USER_API_BASE_URI = "wss://stream.binance.com:9443/ws/";

    public static BinanceUserDataStreamingService create(String listenKey) {
        return new BinanceUserDataStreamingService(USER_API_BASE_URI + listenKey);
    }

    private BinanceUserDataStreamingService(String url) {
        super(url, Integer.MAX_VALUE);
    }

    public Observable<JsonNode> subscribeChannel(BinanceWebSocketTypes eventType) {
    	return super.subscribeChannel(eventType.getSerializedValue());
    }

    @Override
    public void messageHandler(String message) {
        super.messageHandler(message);
    }

    @Override
    protected void handleMessage(JsonNode message) {
        super.handleMessage(message);
    }

    @Override
    protected String getChannelNameFromMessage(JsonNode message) throws IOException {
        return message.get("e").asText();
    }

    @Override
    public String getSubscribeMessage(String channelName, Object... args) throws IOException {
        // No op. Disconnecting from the web socket will cancel subscriptions.
        return null;
    }

    @Override
    public String getUnsubscribeMessage(String channelName) throws IOException {
        // No op. Disconnecting from the web socket will cancel subscriptions.
        return null;
    }

    @Override
    public void sendMessage(String message) {
        // Subscriptions are made upon connection - no messages are sent.
    }
}

@@ -293,24 +293,82 @@
 bservable;%0A%0A
+import org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0A
 import java.
@@ -383,16 +383,16 @@
 eption;%0A
-
 %0Apublic 
@@ -466,16 +466,119 @@
 vice %7B%0A%0A
+    private static final Logger LOG = LoggerFactory.getLogger(BinanceUserDataStreamingService.class);%0A%0A
 %09private
@@ -1141,32 +1141,84 @@
 ring message) %7B%0A
+        LOG.debug(%22Received message: %7B%7D%22, message);%0A
         super.me
@@ -1306,32 +1306,50 @@
 Node message) %7B%0A
+        try %7B%0A    
         super.ha
@@ -1362,32 +1362,158 @@
 ssage(message);%0A
+        %7D catch (Exception e) %7B%0A            LOG.error(%22Error handling message: %22 + message, e);%0A            return;%0A        %7D%0A
     %7D%0A%0A    @Over

Improve logging of BinanceUserDataStreamingService