package info.bitrich.xchangestream.coinbasepro;

import org.knowm.xchange.ExchangeSpecification;
import org.knowm.xchange.coinbasepro.CoinbaseProExchange;
import org.knowm.xchange.coinbasepro.dto.account.CoinbaseProWebsocketAuthData;
import org.knowm.xchange.coinbasepro.service.CoinbaseProAccountServiceRaw;

import info.bitrich.xchangestream.core.ProductSubscription;
import info.bitrich.xchangestream.core.StreamingExchange;
import info.bitrich.xchangestream.core.StreamingMarketDataService;
import info.bitrich.xchangestream.service.netty.WebSocketClientHandler;
import io.reactivex.Completable;
import io.reactivex.Observable;

/**
 * CoinbasePro Streaming Exchange. Connects to live WebSocket feed.
 */
public class CoinbaseProStreamingExchange extends CoinbaseProExchange implements StreamingExchange {
    private static final String API_URI = "wss://ws-feed.pro.coinbase.com";

    private CoinbaseProStreamingService streamingService;
    private CoinbaseProStreamingMarketDataService streamingMarketDataService;

    public CoinbaseProStreamingExchange() { }

    @Override
    protected void initServices() {
        super.initServices();
    }

    @Override
    public Completable connect(ProductSubscription... args) {
        if (args == null || args.length == 0)
            throw new UnsupportedOperationException("The ProductSubscription must be defined!");
        ExchangeSpecification exchangeSpec = getExchangeSpecification();
        this.streamingService = new CoinbaseProStreamingService(API_URI, () -> authData(exchangeSpec));
        this.streamingMarketDataService = new CoinbaseProStreamingMarketDataService(streamingService);
        streamingService.subscribeMultipleCurrencyPairs(args);

        return streamingService.connect();
    }

    private CoinbaseProWebsocketAuthData authData(ExchangeSpecification exchangeSpec) {
        CoinbaseProWebsocketAuthData authData = null;
        if ( exchangeSpec.getApiKey() != null ) {
            try {
                CoinbaseProAccountServiceRaw rawAccountService = (CoinbaseProAccountServiceRaw) getAccountService();
                authData = rawAccountService.getWebsocketAuthData();
            }
            catch (Exception e) {
                logger.warn("Failed attempting to acquire Websocket AuthData needed for private data on" +
                            " websocket.  Will only receive public information via API", e);
            }
        }
        return authData;
    }

    @Override
    public Completable disconnect() {
        CoinbaseProStreamingService service = streamingService;
        streamingService = null;
        streamingMarketDataService = null;
        return service.disconnect();
    }

    @Override
    public Observable<Throwable> reconnectFailure() {
        return streamingService.subscribeReconnectFailure();
    }

    @Override
    public Observable<Object> connectionSuccess() {
        return streamingService.subscribeConnectionSuccess();
    }

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

        return spec;
    }

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

    /**
     * Enables the user to listen on channel inactive events and react appropriately.
     *
     * @param channelInactiveHandler a WebSocketMessageHandler instance.
     */
    public void setChannelInactiveHandler(WebSocketClientHandler.WebSocketMessageHandler channelInactiveHandler) {
        streamingService.setChannelInactiveHandler(channelInactiveHandler);
    }

    @Override
    public boolean isAlive() {
        return streamingService != null && streamingService.isSocketOpen();
    }

    @Override
    public void useCompressedMessages(boolean compressedMessages) { streamingService.useCompressedMessages(compressedMessages); }
}

@@ -878,16 +878,115 @@
 se.com%22;
+%0A    private static final String SANDBOX_API_URI = %22wss://ws-feed-public.sandbox.pro.coinbase.com%22;
 %0A%0A    pr
@@ -1534,32 +1534,193 @@
 pecification();%0A
+        String apiUri = exchangeSpecification.getExchangeSpecificParametersItem(%22Use_Sandbox%22).equals(true)%0A            ? SANDBOX_API_URI%0A            : API_URI;%0A
         this.str
@@ -1767,23 +1767,22 @@
 Service(
-API_URI
+apiUri
 , () -%3E 
@@ -1905,24 +1905,24 @@
 ngService);%0A
+
         stre
@@ -1972,17 +1972,16 @@
 (args);%0A
-%0A
         

Use conventional support for coinbase pro sandbox access