package info.bitrich.xchangestream.okcoin;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.okcoin.dto.WebSocketMessage;
import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
import org.knowm.xchange.exceptions.ExchangeException;

import java.io.IOException;

public class OkCoinStreamingService extends JsonNettyStreamingService {

    public OkCoinStreamingService(String apiUrl) {
        super(apiUrl);
    }

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

    @Override
    public String getSubscribeMessage(String channelName, Object... args) throws IOException {
        WebSocketMessage webSocketMessage = new WebSocketMessage("addChannel", channelName);

        ObjectMapper objectMapper = new ObjectMapper();
        return objectMapper.writeValueAsString(webSocketMessage);
    }

    @Override
    public String getUnsubscribeMessage(String channelName) throws IOException {
        WebSocketMessage webSocketMessage = new WebSocketMessage("removeChannel", channelName);

        ObjectMapper objectMapper = new ObjectMapper();
        return objectMapper.writeValueAsString(webSocketMessage);
    }

    @Override
    protected void handleMessage(JsonNode message) {
        if (message.get("data") != null) {
            if (message.get("data").has("result")) {
                boolean success = message.get("data").get("result").asBoolean();
                if (!success) {
                    super.handleError(message, new ExchangeException("Error code: " + message.get("errorcode").asText()));
                }
                return;
            }
        }
        super.handleMessage(message);
    }
}

@@ -275,16 +275,88 @@
 ervice;%0A
+import io.reactivex.Completable;%0Aimport io.reactivex.CompletableSource;%0A
 import o
@@ -430,16 +430,68 @@
 ception;
+%0Aimport java.util.Timer;%0Aimport java.util.TimerTask;
 %0A%0Apublic
@@ -557,16 +557,57 @@
 vice %7B%0A%0A
+    private Timer pingPongTimer = null;%0A%0A
     publ
@@ -671,32 +671,809 @@
 apiUrl);%0A    %7D%0A%0A
+    @Override%0A    public Completable connect() %7B%0A        Completable conn = super.connect();%0A        return conn.andThen((CompletableSource)(completable) -%3E %7B%0A            try %7B%0A                if (pingPongTimer != null) %7B%0A                    pingPongTimer.cancel();%0A                %7D%0A                pingPongTimer = new Timer(%22OkexPingPong%22, false);%0A                pingPongTimer.schedule(new TimerTask() %7B%0A                    @Override%0A                    public void run() %7B%0A                        OkCoinStreamingService.this.sendMessage(%22%7B%5C%22event%5C%22:%5C%22ping%5C%22%7D%22);%0A                    %7D%0A                %7D, 15*1000, 15*1000);%0A                completable.onComplete();%0A            %7D catch (Exception e) %7B%0A                completable.onError(e);%0A            %7D%0A        %7D);%0A    %7D%0A%0A
     @Override%0A  
@@ -2327,24 +2327,182 @@
  message) %7B%0A
+        if (message.get(%22event%22) != null && %22pong%22.equals(message.get(%22event%22).asText()) ) %7B%0A            // ignore pong message%0A            return;%0A        %7D%0A
         if (

send ping message to server every 15 sec