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 io.reactivex.subjects.PublishSubject;
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.dto.trade.LimitOrder;
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) {
    return PublishSubject
        .create( emitter -> getOrderBook(currencyPair, args)
            .subscribe( orderBook -> {
              LimitOrder firstBid = orderBook.getBids().iterator().next();
              LimitOrder firstAsk = orderBook.getAsks().iterator().next();
              emitter.onNext(new Ticker.Builder()
                  .currencyPair(currencyPair)
                  .bid(firstBid.getLimitPrice())
                  .bidSize(firstBid.getOriginalAmount())
                  .ask(firstAsk.getLimitPrice())
                  .askSize(firstAsk.getOriginalAmount())
                  .timestamp(firstBid.getTimestamp().after(firstAsk.getTimestamp()) ? firstBid.getTimestamp() : firstAsk.getTimestamp())
                  .build());
            }));
  }

  @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());
  }
}

@@ -943,79 +943,8 @@
 er;%0A
-import org.knowm.xchange.exceptions.NotAvailableFromExchangeException;%0A
 impo
@@ -4629,36 +4629,47 @@
 ject
-%0A        .create( emitter -%3E
+.create(%0A        emitter -%3E%0A           
  get
@@ -4702,32 +4702,36 @@
 gs)%0A            
+    
 .subscribe( orde
@@ -4725,16 +4725,36 @@
 bscribe(
+%0A                   
  orderBo
@@ -4757,24 +4757,32 @@
 erBook -%3E %7B%0A
+        
             
@@ -4850,32 +4850,40 @@
 ;%0A              
+     
+   
 LimitOrder first
@@ -4937,24 +4937,32 @@
             
+        
 emitter.onNe
@@ -4964,16 +4964,43 @@
 .onNext(
+%0A                          
 new Tick
@@ -5030,16 +5030,28 @@
         
+     
+       
 .currenc
@@ -5070,16 +5070,28 @@
 cyPair)%0A
+            
         
@@ -5145,24 +5145,36 @@
             
+  
+          
 .bidSize(fir
@@ -5192,32 +5192,44 @@
 iginalAmount())%0A
+            
                 
@@ -5275,24 +5275,36 @@
             
+            
 .askSize(fir
@@ -5348,16 +5348,28 @@
         
+         
+   
 .timesta
@@ -5371,16 +5371,51 @@
 mestamp(
+%0A                                  
 firstBid
@@ -5464,34 +5464,110 @@
 p())
- ? firstBid.getTimestamp()
+%0A                                      ? firstBid.getTimestamp()%0A                                     
  : f
@@ -5608,16 +5608,28 @@
         
+            
 .build()
@@ -5643,17 +5643,63 @@
         
+   
-%7D
+     %7D,%0A                    emitter::onError
 ));%0A  %7D%0A

Implement getTicker using top of orderbook - reformat - cascade errors