package info.bitrich.xchangestream.kraken;

import com.google.common.base.MoreObjects;
import info.bitrich.xchangestream.core.*;
import io.reactivex.Completable;
import io.reactivex.Observable;
import org.apache.commons.lang3.StringUtils;
import org.knowm.xchange.ExchangeSpecification;
import org.knowm.xchange.kraken.KrakenExchange;
import org.knowm.xchange.kraken.service.KrakenAccountServiceRaw;
import org.knowm.xchange.utils.nonce.AtomicLongCurrentTimeIncrementalNonceFactory;
import org.knowm.xchange.utils.nonce.CurrentTimeNonceFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import si.mazi.rescu.SynchronizedValueFactory;

/**
 * @author makarid
 */
public class KrakenStreamingExchange extends KrakenExchange implements StreamingExchange {

    private static final Logger LOG = LoggerFactory.getLogger(KrakenStreamingExchange.class);
    private final static String USE_BETA = "Use_Beta";
    private static final String API_URI = "wss://ws.kraken.com";
    private static final String API_AUTH_URI = "wss://ws-auth.kraken.com";
    private static final String API_FUTURES_URI = "wss://futures.kraken.com";
    private static final String API_BETA_URI = "wss://beta-ws.kraken.com";

    private KrakenStreamingService streamingService, privateStreamingService, futuresStreamingService;
    private KrakenStreamingMarketDataService streamingMarketDataService;
    private KrakenStreamingTradeService streamingTradeService;

    private final SynchronizedValueFactory<Long> nonceFactory = new CurrentTimeNonceFactory();

    public KrakenStreamingExchange() {
    }

    private static String pickUri(boolean isPrivate, boolean useBeta) {
        return useBeta ? API_BETA_URI : isPrivate ? API_AUTH_URI : API_URI;
    }

    @Override
    protected void initServices() {
        super.initServices();
        Boolean useBeta = MoreObjects.firstNonNull((Boolean)exchangeSpecification.getExchangeSpecificParametersItem(USE_BETA), Boolean.FALSE);

        this.streamingService = new KrakenStreamingService(false, pickUri(false,useBeta));
        this.streamingMarketDataService = new KrakenStreamingMarketDataService(streamingService);

        if (StringUtils.isNotEmpty(exchangeSpecification.getApiKey())) {
            this.privateStreamingService = new KrakenStreamingService(true, pickUri(true,useBeta));
//            this.futuresStreamingService = new KrakenStreamingService(true, API_FUTURES_URI);
        }

        KrakenAccountServiceRaw rawKrakenAcctService = (KrakenAccountServiceRaw) getAccountService();

        streamingTradeService = new KrakenStreamingTradeService(privateStreamingService, rawKrakenAcctService);
    }

    @Override
    public Completable connect(ProductSubscription... args) {
        if (privateStreamingService != null)
            return privateStreamingService.connect().mergeWith(streamingService.connect());

        return streamingService.connect();
    }

    @Override
    public Completable disconnect() {
        if (privateStreamingService != null)
            return privateStreamingService.disconnect().mergeWith(streamingService.disconnect());

        return streamingService.disconnect();
    }

    @Override
    public boolean isAlive() {
        return streamingService.isSocketOpen() && (privateStreamingService == null || privateStreamingService.isSocketOpen());
    }
    
    @Override
    public Observable<Object> connectionSuccess() {
        if (privateStreamingService != null)
            return streamingService.subscribeConnectionSuccess().mergeWith(privateStreamingService.subscribeConnectionSuccess());

        return streamingService.subscribeConnectionSuccess();
    }
    
    @Override
    public Observable<Throwable> reconnectFailure() {
        if (privateStreamingService != null)
            return streamingService.subscribeReconnectFailure().mergeWith(privateStreamingService.subscribeReconnectFailure());

        return streamingService.subscribeReconnectFailure();
    }
    
    @Override
    public SynchronizedValueFactory<Long> getNonceFactory() {
        return nonceFactory;
    }

    @Override
    public ExchangeSpecification getDefaultExchangeSpecification() {
        ExchangeSpecification spec = super.getDefaultExchangeSpecification();
        spec.setShouldLoadRemoteMetaData(false);
        return spec;
    }

    @Override
    public StreamingMarketDataService getStreamingMarketDataService() {
        return streamingMarketDataService;
    }

    @Override
    public StreamingTradeService getStreamingTradeService() {
        return streamingTradeService;
    }
    @Override
    public void useCompressedMessages(boolean compressedMessages) {
        streamingService.useCompressedMessages(compressedMessages);
    }
}

@@ -119,17 +119,222 @@
 am.core.
-*
+ProductSubscription;%0Aimport info.bitrich.xchangestream.core.StreamingExchange;%0Aimport info.bitrich.xchangestream.core.StreamingMarketDataService;%0Aimport info.bitrich.xchangestream.core.StreamingTradeService
 ;%0Aimport
@@ -602,91 +602,8 @@
 aw;%0A
-import org.knowm.xchange.utils.nonce.AtomicLongCurrentTimeIncrementalNonceFactory;%0A
 impo
@@ -1205,31 +1205,28 @@
  String API_
-FUTURES
+BETA
 _URI = %22wss:
@@ -1227,22 +1227,22 @@
  %22wss://
-future
+beta-w
 s.kraken
@@ -1240,32 +1240,34 @@
 ws.kraken.com%22;%0A
+//
     private stat
@@ -1278,36 +1278,39 @@
 inal String API_
-BETA
+FUTURES
 _URI = %22wss://be
@@ -1299,38 +1299,38 @@
 ES_URI = %22wss://
-beta-w
+future
 s.kraken.com%22;%0A%0A
@@ -1405,17 +1405,54 @@
 gService
-,
+;%0A//    private KrakenStreamingService
  futures

comment out all references to Futures channel (for now)