package info.bitrich.xchangestream.poloniex2;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.poloniex2.dto.PoloniexWebSocketEvent;
import info.bitrich.xchangestream.poloniex2.dto.PoloniexWebSocketEventsTransaction;
import info.bitrich.xchangestream.poloniex2.dto.PoloniexWebSocketSubscriptionMessage;
import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
import io.reactivex.Completable;
import io.reactivex.Observable;
import org.knowm.xchange.currency.CurrencyPair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

/**
 * Created by Lukas Zaoralek on 10.11.17.
 */
public class PoloniexStreamingService extends JsonNettyStreamingService {
    private static final Logger LOG = LoggerFactory.getLogger(PoloniexStreamingService.class);

    private static final String HEARTBEAT = "1010";

    private final Map<String, String> subscribedChannels = new HashMap<>();
    private final Map<String, Observable<JsonNode>> subscriptions = new HashMap<>();


    public PoloniexStreamingService(String apiUrl) {
        super(apiUrl, Integer.MAX_VALUE);
    }

    @Override
    protected void handleMessage(JsonNode message) {

        if (message.isArray()) {
            if (message.size() < 3) {
                if (message.get(0).asText().equals(HEARTBEAT)) return;
                else if (message.get(0).asText().equals("1002")) return;
            }
            Integer channelId = new Integer(message.get(0).toString());
            if (channelId > 0 && channelId < 1000) {
                JsonNode events = message.get(2);
                if (events.isArray()) {
                    JsonNode event = events.get(0);
                    if (event.get(0).toString().equals("\"i\"")) {
                        if (event.get(1).has("orderBook")) {
                            String currencyPair = event.get(1).get("currencyPair").asText();
                            LOG.info("Register {} as {}", String.valueOf(channelId), currencyPair);
                            subscribedChannels.put(String.valueOf(channelId), currencyPair);
                        }
                    }
                }
            }
        }
        if (message.has("error")) {
            LOG.error("Error with message: " + message.get("error").asText());
            return;
        }
        super.handleMessage(message);
    }

    @Override
    public boolean processArrayMassageSeparately() {
        return false;
    }

    @Override
    public Observable<JsonNode> subscribeChannel(String channelName, Object... args) {
        if (!channels.containsKey(channelName)) {
            Observable<JsonNode> subscription = super.subscribeChannel(channelName, args);
            subscriptions.put(channelName, subscription);
        }

        return subscriptions.get(channelName);
    }

    public Observable<PoloniexWebSocketEvent> subscribeCurrencyPairChannel(CurrencyPair currencyPair) {
        String channelName = currencyPair.counter.toString() + "_" + currencyPair.base.toString();
        return subscribeChannel(channelName)
                .scan(Optional.<JsonNode>empty(), (jsonNodeOldOptional, jsonNodeNew) -> {
                    jsonNodeOldOptional.ifPresent(jsonNode -> {
                        if (jsonNode.get(1).longValue() + 1 != jsonNodeNew.get(1).longValue()) {
                            try {
                                throw new RuntimeException("Invalid sequencing, old: " + jsonNodeOldOptional.map(n -> {
                                    try {
                                        return objectMapper.writeValueAsString(n);
                                    } catch (JsonProcessingException e) {
                                        throw new RuntimeException(e);
                                    }
                                }).orElse("empty") + " new: " + objectMapper.writeValueAsString(jsonNodeNew));
                            } catch (JsonProcessingException e) {
                                throw new RuntimeException(e);
                            }
                        }
                    });
                    return Optional.of(jsonNodeNew);
                })
                .filter(Optional::isPresent)
                .map(Optional::get)
                .flatMapIterable(s -> {
                    PoloniexWebSocketEventsTransaction transaction = objectMapper.treeToValue(s, PoloniexWebSocketEventsTransaction.class);
                    return Arrays.asList(transaction.getEvents());
                }).share();
    }

    @Override
    protected String getChannelNameFromMessage(JsonNode message) throws IOException {
        String strChannelId = message.get(0).asText();
        Integer channelId = new Integer(strChannelId);
        if (channelId >= 1000) return strChannelId;
        else return subscribedChannels.get(message.get(0).asText());
    }

    @Override
    public String getSubscribeMessage(String channelName, Object... args) throws IOException {
        PoloniexWebSocketSubscriptionMessage subscribeMessage = new PoloniexWebSocketSubscriptionMessage("subscribe",
                channelName);

        final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
        return objectMapper.writeValueAsString(subscribeMessage);
    }

    @Override
    public String getUnsubscribeMessage(String channelName) throws IOException {
        PoloniexWebSocketSubscriptionMessage subscribeMessage = new PoloniexWebSocketSubscriptionMessage("unsubscribe",
                channelName);

        final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
        return objectMapper.writeValueAsString(subscribeMessage);
    }

    @Override
    public Completable disconnect() {

        return super.disconnect();
    }

}

@@ -3645,24 +3645,213 @@
 gValue()) %7B%0A
+                            if (jsonNodeNew.get(2).size() == 1 && jsonNodeNew.get(2).get(0).textValue().equals(%22i%22)) %7B%0A                                return;%0A                            %7D%0A
             

Allow "i" messages to break continuous sequence