package info.bitrich.xchangestream.cexio;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.cexio.dto.CexioWebSocketOrderBookSubscribeResponse;
import info.bitrich.xchangestream.core.StreamingMarketDataService;
import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
import io.reactivex.Observable;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Date;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.dto.marketdata.OrderBook;
import org.knowm.xchange.dto.marketdata.Ticker;
import org.knowm.xchange.dto.marketdata.Trade;
import org.knowm.xchange.dto.trade.LimitOrder;
import org.knowm.xchange.exceptions.NotYetImplementedForExchangeException;

public class CexioStreamingMarketDataService implements StreamingMarketDataService {

  private final CexioStreamingRawService streamingOrderDataService;

  public CexioStreamingMarketDataService(CexioStreamingRawService streamingOrderDataService) {
    this.streamingOrderDataService = streamingOrderDataService;
  }

  static class OrderBookUpdateConsumer
      implements io.reactivex.functions.Function<
          CexioWebSocketOrderBookSubscribeResponse, OrderBook> {
    BigInteger prevID = null;
    OrderBook orderBookSoFar =
        new OrderBook(new Date(), new ArrayList<LimitOrder>(), new ArrayList<LimitOrder>());
    final CexioStreamingRawService streamingOrderDataService;

    public OrderBookUpdateConsumer(CexioStreamingRawService streamingOrderDataService) {
      this.streamingOrderDataService = streamingOrderDataService;
    }

    @Override
    public OrderBook apply(CexioWebSocketOrderBookSubscribeResponse t) throws Exception {
      OrderBook retVal;
      if (prevID != null && prevID.add(BigInteger.ONE).compareTo(t.id) != 0) {
        throw new IllegalStateException(
            "Received an update message with id ["
                + t.id
                + "] not sequential to last id ["
                + prevID
                + "]. "
                + "Orderbook out of order!");
      }

      prevID = t.id;
      orderBookSoFar = CexioAdapters.adaptOrderBookIncremental(orderBookSoFar, t);
      retVal = orderBookSoFar;

      return retVal;
    }
  }

  @Override
  public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) {
    String channelNameForPair =
        CexioStreamingRawService.GetOrderBookChannelForCurrencyPair(currencyPair);

    final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
    // check depth parameter
    int depth = 0;
    if (args != null && args[0] instanceof Integer) {
      depth = (Integer) args[0];
    }
    Observable<JsonNode> jsonNodeObservable =
        streamingOrderDataService.subscribeChannel(channelNameForPair, currencyPair, depth);
    OrderBookUpdateConsumer orderBookConsumer =
        new OrderBookUpdateConsumer(streamingOrderDataService);
    return jsonNodeObservable
        .map(
            s -> {
              JsonNode dataNode = s.get("data");
              return mapper.readValue(
                  dataNode.toString(), CexioWebSocketOrderBookSubscribeResponse.class);
            })
        .map(orderBookConsumer);
  }

  @Override
  public Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args) {
    throw new NotYetImplementedForExchangeException();
  }

  @Override
  public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
    throw new NotYetImplementedForExchangeException();
  }
}

@@ -3374,56 +3374,1381 @@
     
-throw new NotYetImplementedForExchangeException(
+String channelNameForPair =%0A            CexioStreamingRawService.GetOrderBookChannelForCurrencyPair(currencyPair);%0A%0A    final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();%0A    int depth = 1;%0A    Observable%3CJsonNode%3E jsonNodeObservable =%0A            streamingOrderDataService.subscribeChannel(channelNameForPair, currencyPair, depth);%0A    OrderBookUpdateConsumer orderBookConsumer =%0A            new OrderBookUpdateConsumer(streamingOrderDataService);%0A%0A    return jsonNodeObservable%0A            .map(%0A                    s -%3E %7B%0A                      JsonNode dataNode = s.get(%22data%22);%0A                      return mapper.readValue(%0A                              dataNode.toString(), CexioWebSocketOrderBookSubscribeResponse.class);%0A                    %7D)%0A            .map(orderBookConsumer)%0A            .filter(orderBook -%3E !orderBook.getBids().isEmpty() && !orderBook.getAsks().isEmpty())%0A            .map(orderBook -%3E new Ticker.Builder()%0A                    .timestamp(orderBook.getTimeStamp())%0A                    .instrument(currencyPair)%0A                    .bid(orderBook.getBids().get(0).getLimitPrice())%0A                    .ask(orderBook.getAsks().get(0).getLimitPrice())%0A                    .build())%0A            .distinctUntilChanged((prev, next) -%3E prev.getBid().compareTo(next.getBid()) == 0 && prev.getAsk().compareTo(next.getAsk()) == 0
 );%0A 

implement streaming ticker