<commit_before>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.ProductSubscription;
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.knowm.xchange.dto.Order;
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.io.IOException;
import java.util.concurrent.*;

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

    private void initWatchDog() {
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new WatchDog(), 10, 10, TimeUnit.SECONDS);
    }

    private void connectAndStart() throws IOException {
        ProductSubscription subscription = ProductSubscription.create()
                .addTrades(CurrencyPair.BTC_USD)
                .build();
        exchange.connect(subscription).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 boolean isHealthy() {
        return !tradeSubscription.isDisposed() && exchange.isAlive();
    }

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

    private Trade getDomainTrade(org.knowm.xchange.dto.marketdata.Trade gdaxTrade) {
        Trade aTrade = new Trade();
        aTrade.setId(gdaxTrade.getId());
        aTrade.setMarket("GDAX");
        aTrade.setCurrency(gdaxTrade.getCurrencyPair().counter.toString());
        aTrade.setTradedAsset(gdaxTrade.getCurrencyPair().base.toString());
        aTrade.setExecutionTime(gdaxTrade.getTimestamp().toInstant());
        aTrade.setPrice(gdaxTrade.getPrice());
        aTrade.setQuantity(gdaxTrade.getOriginalAmount());
        aTrade.setSide(gdaxTrade.getType() == Order.OrderType.BID ? "BUY" : "SELL");

        return aTrade;
    }

    @PreDestroy
    public void cleanupBeforeExit() throws Exception {
        if (tradeSubscription != null && !tradeSubscription.isDisposed()) {
            tradeSubscription.dispose();
        }
<commit_msg>Check for disconnection alert<commit_after>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.ProductSubscription;
import info.bitrich.xchangestream.core.StreamingExchange;
import info.bitrich.xchangestream.core.StreamingExchangeFactory;
import info.bitrich.xchangestream.gdax.GDAXStreamingExchange;
import info.bitrich.xchangestream.service.netty.WebSocketClientHandler;
import io.reactivex.disposables.Disposable;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.dto.Order;
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.io.IOException;
import java.util.concurrent.*;

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

    @Override
    public void run(ApplicationArguments applicationArguments) throws Exception {
        connectAndStart();
        initWatchDog();
    }

    private void initWatchDog() {
        ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
        service.scheduleAtFixedRate(new WatchDog(), 10, 10, TimeUnit.SECONDS);
    }

    private void connectAndStart() throws IOException {
        ProductSubscription subscription = ProductSubscription.create()
                .addTrades(CurrencyPair.BTC_USD)
                .build();
        exchange.connect(subscription).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);
                });

        exchange.setChannelInactiveHandler(new WebSocketClientHandler.WebSocketMessageHandler() {
            @Override
            public void onMessage(String message) {
                isDisconnectedAlert = true;
            }
        });

    }

    private boolean isHealthy() {
        return !tradeSubscription.isDisposed() && exchange.isAlive() && !isDisconnectedAlert;
    }

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

    private Trade getDomainTrade(org.knowm.xchange.dto.marketdata.Trade gdaxTrade) {
        Trade aTrade = new Trade();
        aTrade.setId(gdaxTrade.getId());
        aTrade.setMarket("GDAX");
        aTrade.setCurrency(gdaxTrade.getCurrencyPair().counter.toString());
        aTrade.setTradedAsset(gdaxTrade.getCurrencyPair().base.toString());
        aTrade.setExecutionTime(gdaxTrade.getTimestamp().toInstant());
        aTrade.setPrice(gdaxTrade.getPrice());
        aTrade.setQuantity(gdaxTrade.getOriginalAmount());
        aTrade.setSide(gdaxTrade.getType() == Order.OrderType.BID ? "BUY" : "SELL");

        return aTrade;
    }

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

{'fraction_non_alphanumeric': 0.10102007239223429, 'fraction_numerical': 0.003948667324777887, 'mean_word_length': 2.8982471141513466, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3438894', 'n_tokens_mistral': 2592, 'n_tokens_neox': 2500, 'n_words': 411}