package org.chuggol.crypto.gateway.gdax;

import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import info.bitrich.xchangestream.core.StreamingExchange;
import info.bitrich.xchangestream.core.StreamingExchangeFactory;
import info.bitrich.xchangestream.gdax.GDAXStreamingExchange;
import io.reactivex.disposables.Disposable;
import org.knowm.xchange.currency.CurrencyPair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

@Service
public class GdaxInboundGatewayRunner implements ApplicationRunner {
    private static final Logger LOG = LoggerFactory.getLogger(GdaxInboundGatewayRunner.class);
    private final StreamingExchange exchange = StreamingExchangeFactory.INSTANCE.createExchange(GDAXStreamingExchange.class.getName());
    private Disposable tradeSubscription;
    private Gson gson = new GsonBuilder().create();
    private Publisher publisher;

    @Override
    public void run(ApplicationArguments applicationArguments) throws Exception {
        exchange.connect().blockingAwait();
        TopicName topicName = TopicName.create("crypto-175617", "inbound-trades");
        publisher = Publisher.defaultBuilder(topicName).build();


        tradeSubscription = exchange.getStreamingMarketDataService()
                .getTrades(CurrencyPair.BTC_USD)
                .subscribe(gdaxTrade -> {
                    if (!"0".equals(gdaxTrade.getId())) {
                        Trade trade = getDomainTrade(gdaxTrade);
                        String tradeJson = toJson(trade);
                        ByteString tradeData = ByteString.copyFromUtf8(tradeJson);
                        PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(tradeData).build();
                        LOG.info("Incoming gdaxTrade: {}", tradeJson);
                        ApiFuture<String> future = publisher.publish(pubsubMessage);
                        try {
                            LOG.info("Published with id: {}", future.get());
                        } catch (Exception ex) {
                            LOG.error("Got exception: ", ex);
                        }


                    }
                }, throwable -> {
                    LOG.error("Error in subscribing trades.", throwable);
                });
    }

    private String toJson(Trade trade) {
        return gson.toJson(trade);
    }

    private Trade getDomainTrade(org.knowm.xchange.dto.marketdata.Trade gdaxTrade) {
        Trade aTrade = new Trade();
        aTrade.setCurrencyPair(gdaxTrade.getCurrencyPair().toString());
        aTrade.setId(gdaxTrade.getId());
        aTrade.setPrice(gdaxTrade.getPrice());
        aTrade.setQuantity(gdaxTrade.getTradableAmount());
        aTrade.setId(gdaxTrade.getId());
        aTrade.setSide(gdaxTrade.getType().name());
        return aTrade;
    }

    @PreDestroy
    public void cleanupBeforeExit() throws Exception {
        if (tradeSubscription != null && !tradeSubscription.isDisposed()) {
            tradeSubscription.dispose();
        }

        exchange.disconnect().blockingAwait();

        publisher.shutdown();
    }
}

@@ -843,78 +843,22 @@
 ava.
-util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors
+io.IOException
 ;%0Aim
@@ -887,26 +887,9 @@
 ent.
-ThreadPoolExecutor
+*
 ;%0A%0A@
@@ -1409,32 +1409,345 @@
 throws Exception
+%7B%0A        connectAndStart();%0A        initWatchDog();%0A    %7D%0A%0A    private void initWatchDog() %7B%0A        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);%0A        service.scheduleAtFixedRate(new WatchDog(), 10, 10, TimeUnit.SECONDS);%0A    %7D%0A%0A    private void connectAndStart() throws IOException
  %7B%0A        excha
@@ -3011,16 +3011,106 @@
      %7D);
+%0A%0A    %7D%0A%0A    private boolean isHealthy() %7B%0A        return !tradeSubscription.isDisposed();
 %0A    %7D%0A%0A
@@ -3898,24 +3898,24 @@
 ngAwait();%0A%0A
-
         publ
@@ -3937,11 +3937,603 @@
 );%0A    %7D
+%0A%0A    class WatchDog implements Runnable %7B%0A%0A        @Override%0A        public void run() %7B%0A            LOG.info(%22Checking subscription health%22);%0A            if (!isHealthy()) %7B%0A                try %7B%0A                    cleanupBeforeExit();%0A                %7D catch (Exception ex) %7B%0A                    LOG.warn(%22Caught exception during restart/cleanup%22, ex);%0A                %7D%0A%0A                try %7B%0A                    connectAndStart();%0A                %7D catch (Exception ex) %7B%0A                    LOG.warn(%22Caught exception during start%22, ex);%0A                %7D%0A            %7D%0A        %7D%0A    %7D
 %0A%7D%0A

Add subscription health watchdog