support authorized channels

pom.xml

@@ -5,7 +5,7 @@
     <groupId>info.bitrich.xchange-stream</groupId>
     <artifactId>xchange-stream-parent</artifactId>
     <packaging>pom</packaging>
-    <version>4.4.0.3-dsx-SNAPSHOT</version>
+    <version>4.4.0.4-dsx-SNAPSHOT</version>
 
     <modules>
         <module>xchange-stream-core</module>

service-core/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

service-netty/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

service-netty/src/main/java/info/bitrich/xchangestream/service/netty/StreamingObjectMapperHelper.java

@@ -19,6 +19,7 @@ public class StreamingObjectMapperHelper {
         objectMapper.enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
         objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
         objectMapper.setSerializationInclusion(JsonInclude.Include.NON_DEFAULT);
+        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_EMPTY);
     }
 
     private StreamingObjectMapperHelper() {

service-pubnub/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

service-pusher/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

service-wamp/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-bankera/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-binance/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-bitfinex/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-bitflyer/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-bitmex/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-bitstamp/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-cexio/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-coinbasepro/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-coinmate/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-dsx/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>xchange-stream-parent</artifactId>
         <groupId>info.bitrich.xchange-stream</groupId>
-        <version>4.4.0.3-dsx-SNAPSHOT</version>
+        <version>4.4.0.4-dsx-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

xchange-dsx/src/main/java/info/bitrich/xchangestream/dsx/DsxStreamingAccountService.java

@@ -0,0 +1,42 @@
+package info.bitrich.xchangestream.dsx;
+
+import info.bitrich.xchangestream.core.StreamingAccountService;
+import info.bitrich.xchangestream.dsx.dto.messages.DsxAuthBalanceMessage;
+import io.reactivex.Observable;
+import org.knowm.xchange.currency.Currency;
+import org.knowm.xchange.dto.account.Balance;
+import org.knowm.xchange.exceptions.ExchangeException;
+import org.knowm.xchange.exceptions.ExchangeSecurityException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class DsxStreamingAccountService implements StreamingAccountService {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DsxStreamingAccountService.class);
+
+    private final DsxStreamingService service;
+
+    public DsxStreamingAccountService(DsxStreamingService service) {
+        this.service = service;
+    }
+
+    @Override
+    public Observable<Balance> getBalanceChanges(Currency currency, Object... args) {
+        if (args.length == 0 || !Integer.class.isInstance(args[0])) {
+            throw new ExchangeException("Specify faId to monitor balance stream");
+        }
+        Integer faId = (Integer) args[0];
+        return getRawAuthenticatedBalances()
+                .filter(b -> b.getFaId().equals(faId))
+                .filter(b -> currency.getCurrencyCode().equals(b.getCurrency()))
+                .map(DsxStreamingMessageAdapter::adaptBalance);
+    }
+
+    public Observable<DsxAuthBalanceMessage> getRawAuthenticatedBalances() {
+        if (!service.isAuthDataProvided()) {
+            throw new ExchangeSecurityException("Not authenticated");
+        }
+        return service.getAuthenticatedBalances();
+    }
+}

xchange-dsx/src/main/java/info/bitrich/xchangestream/dsx/DsxStreamingExchange.java

@@ -5,8 +5,10 @@
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import io.reactivex.Completable;
 import io.reactivex.Observable;
+import org.apache.commons.lang3.StringUtils;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.dsx.DSXExchange;
+import org.knowm.xchange.exceptions.ExchangeSecurityException;
 
 import static info.bitrich.xchangestream.service.ConnectableService.BEFORE_CONNECTION_HANDLER;
 
@@ -19,6 +21,8 @@ public class DsxStreamingExchange extends DSXExchange implements StreamingExchan
 
     private DsxStreamingService streamingService;
     private DsxStreamingMarketDataService streamingMarketDataService;
+    private DsxStreamingTradeService streamingTradeService;
+    private DsxStreamingAccountService streamingAccountService;
 
     public DsxStreamingExchange() {
     }
@@ -27,9 +31,21 @@ public DsxStreamingExchange() {
     protected void initServices() {
         super.initServices();
         Object apiURI = getExchangeSpecification().getExchangeSpecificParametersItem(DSX_SPEC_PARAMS_API_URI);
-        streamingService = new DsxStreamingService(apiURI == null ? DEFAULT_API_URI : (String) apiURI);
+        streamingService = createStreamingService(apiURI == null ? DEFAULT_API_URI : (String) apiURI);
         streamingService.setBeforeConnectionHandler((Runnable) getExchangeSpecification().getExchangeSpecificParametersItem(BEFORE_CONNECTION_HANDLER));
         streamingMarketDataService = new DsxStreamingMarketDataService(streamingService);
+        streamingTradeService = new DsxStreamingTradeService(streamingService);
+        streamingAccountService = new DsxStreamingAccountService(streamingService);
+    }
+
+    private DsxStreamingService createStreamingService(String apiURI) {
+        DsxStreamingService streamingService = new DsxStreamingService(apiURI, getNonceFactory());
+        applyStreamingSpecification(getExchangeSpecification(), streamingService);
+        if (StringUtils.isNotEmpty(exchangeSpecification.getApiKey())) {
+            streamingService.setApiKey(exchangeSpecification.getApiKey());
+            streamingService.setApiSecret(exchangeSpecification.getSecretKey());
+        }
+        return streamingService;
     }
 
     @Override
@@ -69,6 +85,31 @@ public StreamingMarketDataService getStreamingMarketDataService() {
         return streamingMarketDataService;
     }
 
+    public void authorize() {
+        if (!streamingService.isAuthDataProvided()) {
+            throw new ExchangeSecurityException("authenticated data not provided");
+        }
+        if (!streamingService.isAuthorized()) {
+            streamingService.authorize();
+        }
+    }
+
+    @Override
+    public DsxStreamingTradeService getStreamingTradeService() {
+        if (!streamingService.isAuthDataProvided()) {
+            throw new ExchangeSecurityException("No authentication data provided");
+        }
+        return streamingTradeService;
+    }
+
+    @Override
+    public DsxStreamingAccountService getStreamingAccountService() {
+        if (!streamingService.isAuthDataProvided()) {
+            throw new ExchangeSecurityException("No authentication data provided");
+        }
+        return streamingAccountService;
+    }
+
     @Override
     public void useCompressedMessages(boolean compressedMessages) {
         streamingService.useCompressedMessages(compressedMessages);

xchange-dsx/src/main/java/info/bitrich/xchangestream/dsx/DsxStreamingMarketDataService.java

@@ -46,7 +46,7 @@ public DsxStreamingMarketDataService(DsxStreamingService service) {
 
     @Override
     public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) {
-        String channelName = createChannelName(DsxChannel.book, currencyPair, args);
+        String channelName = createChannelName(DsxChannel.BOOK, currencyPair, args);
         final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
 
         Observable<JsonNode> jsonNodeObservable = service.subscribeChannel(channelName, args);
@@ -63,7 +63,7 @@ public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... a
 
     @Override
     public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
-        String channelName = createChannelName(DsxChannel.trade, currencyPair, args);
+        String channelName = createChannelName(DsxChannel.TRADE, currencyPair, args);
         final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
         Observable<JsonNode> jsonNodeObservable = service.subscribeChannel(channelName, args);
         return jsonNodeObservable
@@ -82,7 +82,7 @@ public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) {
 
     @Override
     public Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args) {
-        String channelName = createChannelName(DsxChannel.ticker, currencyPair, args);
+        String channelName = createChannelName(DsxChannel.TICKER, currencyPair, args);
         final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper();
         Observable<JsonNode> jsonNodeObservable = service.subscribeChannel(channelName);
         return jsonNodeObservable