<commit_before>package info.bitrich.xchangestream.binance;

import info.bitrich.xchangestream.binance.BinanceUserDataChannel.NoActiveChannelException;
import info.bitrich.xchangestream.core.ProductSubscription;
import info.bitrich.xchangestream.core.StreamingExchange;

import io.reactivex.Completable;
import io.reactivex.Observable;

import si.mazi.rescu.RestProxyFactory;

import org.knowm.xchange.binance.BinanceAuthenticated;
import org.knowm.xchange.binance.BinanceExchange;
import org.knowm.xchange.binance.service.BinanceMarketDataService;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.service.BaseExchangeService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class BinanceStreamingExchange extends BinanceExchange implements StreamingExchange {

    private static final Logger LOG = LoggerFactory.getLogger(BinanceStreamingExchange.class);
    private static final String API_BASE_URI = "wss://stream.binance.com:9443/";

    private BinanceStreamingService streamingService;
    private BinanceUserDataStreamingService userDataStreamingService;

    private BinanceStreamingMarketDataService streamingMarketDataService;
    private BinanceStreamingAccountService streamingAccountService;
    private BinanceStreamingTradeService streamingTradeService;

    private BinanceUserDataChannel userDataChannel;

    public BinanceStreamingExchange() {
    }

    /**
     * Binance streaming API expects connections to multiple channels to be defined at connection time. To define the channels for this
     * connection pass a `ProductSubscription` in at connection time.
     *
     * @param args A single `ProductSubscription` to define the subscriptions required to be available during this connection.
     * @return
     */
    @Override
    public Completable connect(ProductSubscription... args) {
        if (args == null || args.length == 0) {
            throw new IllegalArgumentException("Subscriptions must be made at connection time");
        }
        if (streamingService != null) {
            throw new UnsupportedOperationException("Exchange only handles a single connection - disconnect the current connection.");
        }

        ProductSubscription subscriptions = args[0];
        streamingService = createStreamingService(subscriptions);

        List<Completable> completables = new ArrayList<>();

        if (!subscriptions.isEmpty()) {
            completables.add(streamingService.connect());
        }

        if (exchangeSpecification.getApiKey() != null) {
            LOG.info("Connecting to authenticated web socket");
            BinanceAuthenticated binance = RestProxyFactory.createProxy(
                BinanceAuthenticated.class,
                getExchangeSpecification().getSslUri(),
                new BaseExchangeService<BinanceExchange>(this) {}.getClientConfig()
            );
            userDataChannel = new BinanceUserDataChannel(binance, exchangeSpecification.getApiKey());
            try {
                completables.add(createAndConnectUserDataService(userDataChannel.getListenKey()));
            } catch (NoActiveChannelException e) {
                throw new IllegalStateException("Failed to establish user data channel", e);
            }
        }

        streamingMarketDataService = new BinanceStreamingMarketDataService(streamingService, (BinanceMarketDataService) marketDataService);
        streamingAccountService = new BinanceStreamingAccountService(userDataStreamingService);
        streamingTradeService = new BinanceStreamingTradeService(userDataStreamingService);


        return Completable.concat(completables)
            .doOnComplete(() -> streamingMarketDataService.openSubscriptions(subscriptions));
    }

    private Completable createAndConnectUserDataService(String listenKey) {<commit_msg>Re-enable authenticated streams, disabled by accident After refactoring to move account and trade streams to separate classes, I forgot to subscribe those classes on connection.<commit_after>package info.bitrich.xchangestream.binance;

import info.bitrich.xchangestream.binance.BinanceUserDataChannel.NoActiveChannelException;
import info.bitrich.xchangestream.core.ProductSubscription;
import info.bitrich.xchangestream.core.StreamingExchange;

import io.reactivex.Completable;
import io.reactivex.Observable;

import si.mazi.rescu.RestProxyFactory;

import org.knowm.xchange.binance.BinanceAuthenticated;
import org.knowm.xchange.binance.BinanceExchange;
import org.knowm.xchange.binance.service.BinanceMarketDataService;
import org.knowm.xchange.currency.CurrencyPair;
import org.knowm.xchange.service.BaseExchangeService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class BinanceStreamingExchange extends BinanceExchange implements StreamingExchange {

    private static final Logger LOG = LoggerFactory.getLogger(BinanceStreamingExchange.class);
    private static final String API_BASE_URI = "wss://stream.binance.com:9443/";

    private BinanceStreamingService streamingService;
    private BinanceUserDataStreamingService userDataStreamingService;

    private BinanceStreamingMarketDataService streamingMarketDataService;
    private BinanceStreamingAccountService streamingAccountService;
    private BinanceStreamingTradeService streamingTradeService;

    private BinanceUserDataChannel userDataChannel;

    public BinanceStreamingExchange() {
    }

    /**
     * Binance streaming API expects connections to multiple channels to be defined at connection time. To define the channels for this
     * connection pass a `ProductSubscription` in at connection time.
     *
     * @param args A single `ProductSubscription` to define the subscriptions required to be available during this connection.
     * @return
     */
    @Override
    public Completable connect(ProductSubscription... args) {
        if (args == null || args.length == 0) {
            throw new IllegalArgumentException("Subscriptions must be made at connection time");
        }
        if (streamingService != null) {
            throw new UnsupportedOperationException("Exchange only handles a single connection - disconnect the current connection.");
        }

        ProductSubscription subscriptions = args[0];
        streamingService = createStreamingService(subscriptions);

        List<Completable> completables = new ArrayList<>();

        if (!subscriptions.isEmpty()) {
            completables.add(streamingService.connect());
        }

        if (exchangeSpecification.getApiKey() != null) {
            LOG.info("Connecting to authenticated web socket");
            BinanceAuthenticated binance = RestProxyFactory.createProxy(
                BinanceAuthenticated.class,
                getExchangeSpecification().getSslUri(),
                new BaseExchangeService<BinanceExchange>(this) {}.getClientConfig()
            );
            userDataChannel = new BinanceUserDataChannel(binance, exchangeSpecification.getApiKey());
            try {
                completables.add(createAndConnectUserDataService(userDataChannel.getListenKey()));
            } catch (NoActiveChannelException e) {
                throw new IllegalStateException("Failed to establish user data channel", e);
            }
        }

        streamingMarketDataService = new BinanceStreamingMarketDataService(streamingService, (BinanceMarketDataService) marketDataService);
        streamingAccountService = new BinanceStreamingAccountService(userDataStreamingService);
        streamingTradeService = new BinanceStreamingTradeService(userDataStreamingService);


        return Completable.concat(completables)
            .doOnComplete(() -> streamingMarketDataService.openSubscriptions(subscriptions))
            .doOnComplete(() -> streamingAccountService.openSubscriptions())
            .doOnComplete(() -> streamingTradeService.openSubscriptions());
    }

    private Completable createAndConnectUserDataService(String listenKey) {
{'fraction_non_alphanumeric': 0.07389402041808459, 'fraction_numerical': 0.0019445794846864365, 'mean_word_length': 3.9782214156079854, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, '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': '5532752', 'n_tokens_mistral': 2099, 'n_tokens_neox': 2022, 'n_words': 466}