package info.bitrich.xchangestream.gemini;

import static org.knowm.xchange.gemini.v1.GeminiAdapters.adaptTrades;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import info.bitrich.xchangestream.core.StreamingMarketDataService;
import info.bitrich.xchangestream.gemini.dto.GeminiLimitOrder;
import info.bitrich.xchangestream.gemini.dto.GeminiOrderbook;
import info.bitrich.xchangestream.gemini.dto.GeminiWebSocketTransaction;
import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
import io.reactivex.Observable;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
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.exceptions.NotAvailableFromExchangeException;
import org.knowm.xchange.exceptions.NotYetImplementedForExchangeException;
import org.knowm.xchange.gemini.v1.dto.marketdata.GeminiTrade;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Created by Lukas Zaoralek on 15.11.17. */
public class GeminiStreamingMarketDataService implements StreamingMarketDataService {
  private static final Logger LOG = LoggerFactory.getLogger(GeminiStreamingMarketDataService.class);

  private final GeminiStreamingService service;
  private final Map<CurrencyPair, GeminiOrderbook> orderbooks = new HashMap<>();

  private final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();

  public GeminiStreamingMarketDataService(GeminiStreamingService service) {
    this.service = service;
  }

  private boolean filterEventsByReason(JsonNode message, String type, String reason) {
    boolean hasEvents = false;
    if (message.has("events")) {
      for (JsonNode jsonEvent : message.get("events")) {
        boolean reasonResult =
            reason == null
                || (jsonEvent.has("reason") && jsonEvent.get("reason").asText().equals(reason));
        if (jsonEvent.get("type").asText().equals(type) && reasonResult) {
          hasEvents = true;
          break;
        }
      }
    }

    return hasEvents;
  }

  @Override
  public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) {

    Integer maxDepth = args.length > 0 ? (Integer) args[0] : null;

    Observable<GeminiOrderbook> subscribedOrderbookSnapshot =
        service
            .subscribeChannel(currencyPair, maxDepth, maxDepth)
            .filter(
                s ->
                    filterEventsByReason(s, "change", "initial")
                        || filterEventsByReason(s, "change", "place")
                        || filterEventsByReason(s, "change", "cancel")
                        || filterEventsByReason(s, "change", "trade"))
            .filter(
                s -> // filter out updates that arrive before initial book
                orderbooks.get(currencyPair) != null
                        || filterEventsByReason(s, "change", "initial"))
            .map(
                (JsonNode s) -> {
                  if (filterEventsByReason(s, "change", "initial")) {
                    GeminiWebSocketTransaction transaction =
                        mapper.treeToValue(s, GeminiWebSocketTransaction.class);
                    GeminiOrderbook orderbook = transaction.toGeminiOrderbook(currencyPair);
                    orderbooks.put(currencyPair, orderbook);
                    return orderbook;
                  }

                  if (filterEventsByReason(s, "change", "place")
                      || filterEventsByReason(s, "change", "cancel")
                      || filterEventsByReason(s, "change", "trade")) {

                    GeminiWebSocketTransaction transaction =
                        mapper.treeToValue(s, GeminiWebSocketTransaction.class);
                    GeminiLimitOrder[] levels = transaction.toGeminiLimitOrdersUpdate();
                    GeminiOrderbook orderbook = orderbooks.get(currencyPair);
                    orderbook.updateLevels(levels);
                    return orderbook;
                  }

                  throw new NotYetImplementedForExchangeException(
                      " Unknown message type, even after filtering: " + s.toString());
                });

    return subscribedOrderbookSnapshot.map(
        geminiOrderbook -> GeminiAdaptersX.toOrderbook(geminiOrderbook, maxDepth, new Date()));
  }

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

  @Override
  public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
    Observable<GeminiTrade[]> subscribedTrades =
        service
            .subscribeChannel(currencyPair, args)
            .filter(s -> filterEventsByReason(s, "trade", null))
            .map(
                (JsonNode s) -> {
                  GeminiWebSocketTransaction transaction =
                      mapper.treeToValue(s, GeminiWebSocketTransaction.class);
                  return transaction.toGeminiTrades();
                });

    return subscribedTrades.flatMapIterable(s -> adaptTrades(s, currencyPair).getTrades());
  }
}

@@ -582,16 +582,61 @@
 rvable;%0A
+import io.reactivex.subjects.PublishSubject;%0A
 import j
@@ -888,24 +888,71 @@
 data.Trade;%0A
+import org.knowm.xchange.dto.trade.LimitOrder;%0A
 import org.k
@@ -4679,52 +4679,768 @@
     
-throw new NotAvailableFromExchangeException(
+return PublishSubject%0A        .create( emitter -%3E getOrderBook(currencyPair, args)%0A            .subscribe( orderBook -%3E %7B%0A              LimitOrder firstBid = orderBook.getBids().iterator().next();%0A              LimitOrder firstAsk = orderBook.getAsks().iterator().next();%0A              emitter.onNext(new Ticker.Builder()%0A                  .currencyPair(currencyPair)%0A                  .bid(firstBid.getLimitPrice())%0A                  .bidSize(firstBid.getOriginalAmount())%0A                  .ask(firstAsk.getLimitPrice())%0A                  .askSize(firstAsk.getOriginalAmount())%0A                  .timestamp(firstBid.getTimestamp().after(firstAsk.getTimestamp()) ? firstBid.getTimestamp() : firstAsk.getTimestamp())%0A                  .build());%0A            %7D)
 );%0A 

Implement getTicker using top of orderbook