[XChange] Add initial support for account management

Resolves CAMEL-12170

components/camel-xchange/src/main/docs/xchange-component.adoc

@@ -6,8 +6,7 @@
 The *xchange:* component uses the https://knowm.org/open-source/xchange/[XChange] Java library to provide access to 60+ Bitcoin and Altcoin exchanges.
 It comes with a consistent interface for trading and accessing market data. 
 
-Camel can poll for updates of current market data. It can also be used to query historical data based
-on the parameters defined on the endpoint.
+Camel can get crypto currency market data, query historical data, place market orders and much more.
 
 Maven users will need to add the following dependency to their `pom.xml`
 for this component:
@@ -35,8 +34,6 @@ xchange://exchange?options
 The XChange component has no options.
 // component options: END
 
-
-
 // endpoint options: START
 The XChange endpoint is configured using URI syntax:
 
@@ -67,9 +64,25 @@ with the following path and query parameters:
 |===
 // endpoint options: END
 
-### Exchange data format
+### Authentication
 
-[TODO]
+This component communicates with supported crypto currency exchanges via REST API. Some API requests use simple unauthenticated GET request. 
+For most of the interesting stuff however, you'd need an account with the exchange and have API access keys enabled.
+
+These API access keys need to be guarded tightly, especially so when they also allow for the withdraw functionality. 
+In which case, anyone who can get hold of your API keys can easily transfer funds from your account to some other address i.e. steal your money.   
+
+Your API access keys can be strored in an exchange specific properties file in your SSH directory.
+For Binance for example this would be: `~/.ssh/binance-secret.keys`
+
+----
+##
+# This file MUST NEVER be commited to source control. 
+# It is therefore added to .gitignore. 
+#
+apiKey = GuRW0*********
+secretKey = nKLki************
+----
 
 ### Message Headers
 
@@ -81,5 +94,5 @@ In this sample we find the current Bitcoin market price in USDT:
 
 [source,java]
 ---------------------------------------------------------------------------------------------
-from("xchange:binance?method=ticker&currencyPair=BTC/USDT").to("jms:queue:btc");
+from("direct:ticker").to("xchange:binance?service=market&method=ticker&currencyPair=BTC/USDT")
 ---------------------------------------------------------------------------------------------

components/camel-xchange/src/main/java/org/apache/camel/component/xchange/XChangeAccountProducer.java

@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.xchange;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.xchange.XChangeConfiguration.XChangeMethod;
+import org.apache.camel.impl.DefaultProducer;
+
+public class XChangeAccountProducer extends DefaultProducer {
+
+    public XChangeAccountProducer(XChangeEndpoint endpoint) {
+        super(endpoint);
+    }
+
+    @Override
+    public XChangeEndpoint getEndpoint() {
+        return (XChangeEndpoint) super.getEndpoint();
+    }
+
+    @Override
+    public void process(Exchange exchange) throws Exception {
+
+        XChangeEndpoint endpoint = getEndpoint();
+        XChangeMethod method = endpoint.getConfiguration().getMethod();
+
+        if (XChangeMethod.balances == method) {
+            Object body = endpoint.getBalances();
+            exchange.getMessage().setBody(body);
+        } 
+
+        else if (XChangeMethod.fundingHistory == method) {
+            Object body = endpoint.getFundingHistory();
+            exchange.getMessage().setBody(body);
+        } 
+
+        else if (XChangeMethod.wallets == method) {
+            Object body = endpoint.getWallets();
+            exchange.getMessage().setBody(body);
+        } 
+    }
+}

components/camel-xchange/src/main/java/org/apache/camel/component/xchange/XChangeComponent.java

@@ -26,6 +26,8 @@
 
 public class XChangeComponent extends DefaultComponent {
 
+    private XChange exchange;
+
     @Override
     protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
 
@@ -36,15 +38,29 @@ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Obje
         // Set the the required name of the exchange
         configuration.setName(remaining);
 
-        // Get the XChange implementation
-        Class<? extends Exchange> exchangeClass = configuration.getXChangeClass();
-        Assert.notNull(exchangeClass, "XChange not supported: " + configuration.getName());
-
-        // Create the XChange and associated Endpoint
-        XChange exchange = new XChange(ExchangeFactory.INSTANCE.createExchange(exchangeClass));
+        XChange exchange = createXChange(configuration);
         XChangeEndpoint endpoint = new XChangeEndpoint(uri, this, configuration, exchange);
 
         return endpoint;
     }
 
+    public XChange getXChange() {
+        return exchange;
+    }
+
+    private synchronized XChange createXChange(XChangeConfiguration configuration) {
+
+        if (exchange == null) {
+
+            // Get the XChange implementation
+            Class<? extends Exchange> exchangeClass = configuration.getXChangeClass();
+            Assert.notNull(exchangeClass, "XChange not supported: " + configuration.getName());
+
+            // Create the XChange and associated Endpoint
+            exchange = new XChange(ExchangeFactory.INSTANCE.createExchange(exchangeClass));
+        }
+
+        return exchange;
+    }
+
 }

components/camel-xchange/src/main/java/org/apache/camel/component/xchange/XChangeConfiguration.java

@@ -33,8 +33,18 @@
 @UriParams
 public class XChangeConfiguration {
 
-    public enum XChangeService { marketdata, metadata }
-    public enum XChangeMethod { currencies, currencyMetaData, currencyPairs, currencyPairMetaData, ticker }
+    // Available service
+    public enum XChangeService { marketdata, metadata, account }
+
+    // Available methods
+    public enum XChangeMethod {
+        // Account service methods
+        balances, fundingHistory, wallets, 
+        // Metadata service methods
+        currencies, currencyMetaData, currencyPairs, currencyPairMetaData,
+        // Marketdata service methods
+        ticker 
+        }
 
     public static final String HEADER_CURRENCY = "Currency";
     public static final String HEADER_CURRENCY_PAIR = "CurrencyPair";

components/camel-xchange/src/main/java/org/apache/camel/component/xchange/XChangeEndpoint.java

@@ -16,6 +16,9 @@
  */
 package org.apache.camel.component.xchange;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.stream.Collectors;
 
@@ -28,24 +31,37 @@
 import org.apache.camel.spi.UriParam;
 import org.knowm.xchange.currency.Currency;
 import org.knowm.xchange.currency.CurrencyPair;
+import org.knowm.xchange.dto.account.AccountInfo;
+import org.knowm.xchange.dto.account.Balance;
+import org.knowm.xchange.dto.account.FundingRecord;
+import org.knowm.xchange.dto.account.Wallet;
+import org.knowm.xchange.dto.marketdata.Ticker;
 import org.knowm.xchange.dto.meta.CurrencyMetaData;
 import org.knowm.xchange.dto.meta.CurrencyPairMetaData;
 import org.knowm.xchange.dto.meta.ExchangeMetaData;
+import org.knowm.xchange.service.account.AccountService;
+import org.knowm.xchange.service.marketdata.MarketDataService;
+import org.knowm.xchange.service.trade.params.TradeHistoryParams;
 import org.knowm.xchange.utils.Assert;
 
 @UriEndpoint(firstVersion = "2.21.0", scheme = "xchange", title = "XChange", syntax = "xchange:name", producerOnly = true, label = "blockchain")
 public class XChangeEndpoint extends DefaultEndpoint {
 
     @UriParam
-    private XChangeConfiguration configuration;
+    private final XChangeConfiguration configuration;
     private final XChange exchange;
 
-    public XChangeEndpoint(String uri, XChangeComponent component, XChangeConfiguration properties, XChange exchange) {
+    public XChangeEndpoint(String uri, XChangeComponent component, XChangeConfiguration configuration, XChange exchange) {
         super(uri, component);
-        this.configuration = properties;
+        this.configuration = configuration;
         this.exchange = exchange;
     }
 
+    @Override
+    public XChangeComponent getComponent() {
+        return (XChangeComponent) super.getComponent();
+    }
+
     @Override
     public Consumer createConsumer(Processor processor) throws Exception {
         throw new UnsupportedOperationException();
@@ -57,10 +73,12 @@ public Producer createProducer() throws Exception {
         Producer producer = null;
 
         XChangeService service = getConfiguration().getService();
-        if (XChangeService.metadata == service) {
-            producer = new XChangeMetaDataProducer(this);
+        if (XChangeService.account == service) {
+            producer = new XChangeAccountProducer(this);
         } else if (XChangeService.marketdata == service) {
             producer = new XChangeMarketDataProducer(this);
+        } else if (XChangeService.metadata == service) {
+            producer = new XChangeMetaDataProducer(this);
         }
 
         Assert.notNull(producer, "Unsupported service: " + service);
@@ -76,10 +94,6 @@ public XChangeConfiguration getConfiguration() {
         return configuration;
     }
 
-    public XChange getXChange() {
-        return exchange;
-    }
-
     public List<Currency> getCurrencies() {
         ExchangeMetaData metaData = exchange.getExchangeMetaData();
         return metaData.getCurrencies().keySet().stream().sorted().collect(Collectors.toList());
@@ -101,4 +115,53 @@ public CurrencyPairMetaData getCurrencyPairMetaData(CurrencyPair pair) {
         ExchangeMetaData metaData = exchange.getExchangeMetaData();
         return metaData.getCurrencyPairs().get(pair);
     }
+
+    public List<Balance> getBalances() throws IOException {
+        List<Balance> balances = new ArrayList<>();
+        getWallets().stream().forEach(w -> {
+            for (Balance aux : w.getBalances().values()) {
+                Currency curr = aux.getCurrency();
+                CurrencyMetaData metaData = getCurrencyMetaData(curr);
+                if (metaData != null) {
+                    int scale = metaData.getScale();
+                    double total = aux.getTotal().doubleValue();
+                    double scaledTotal = total * Math.pow(10, scale / 2); 
+                    if (1 <= scaledTotal) {
+                        balances.add(aux);
+                    }
+                }
+            }
+        });
+        return balances.stream().sorted(new Comparator<Balance>() {
+            public int compare(Balance o1, Balance o2) {
+               return o1.getCurrency().compareTo(o2.getCurrency());
+            }
+        }).collect(Collectors.toList());
+    }
+
+    public List<FundingRecord> getFundingHistory() throws IOException {
+        AccountService accountService = exchange.getAccountService();
+        TradeHistoryParams fundingHistoryParams = accountService.createFundingHistoryParams();
+        return accountService.getFundingHistory(fundingHistoryParams).stream().sorted(new Comparator<FundingRecord>() {
+            public int compare(FundingRecord o1, FundingRecord o2) {
+                return o1.getDate().compareTo(o2.getDate());
+            }
+        }).collect(Collectors.toList());
+    }
+
+    public List<Wallet> getWallets() throws IOException {
+        AccountService accountService = exchange.getAccountService();
+        AccountInfo accountInfo = accountService.getAccountInfo();
+        return accountInfo.getWallets().values().stream().sorted(new Comparator<Wallet>() {
+            public int compare(Wallet o1, Wallet o2) {
+                return o1.getName().compareTo(o2.getName());
+            }
+        }).collect(Collectors.toList());
+    }
+
+    public Ticker getTicker(CurrencyPair pair) throws IOException {
+        Assert.notNull(pair, "Null currency pair");
+        MarketDataService marketService = exchange.getMarketDataService();
+        return marketService.getTicker(pair);
+    }
 }

components/camel-xchange/src/main/java/org/apache/camel/component/xchange/XChangeMarketDataProducer.java

@@ -23,15 +23,11 @@
 import org.apache.camel.impl.DefaultProducer;
 import org.knowm.xchange.currency.CurrencyPair;
 import org.knowm.xchange.dto.marketdata.Ticker;
-import org.knowm.xchange.service.marketdata.MarketDataService;
 
 public class XChangeMarketDataProducer extends DefaultProducer {
 
-    private final MarketDataService marketService;
-
     public XChangeMarketDataProducer(XChangeEndpoint endpoint) {
         super(endpoint);
-        marketService = endpoint.getXChange().getMarketDataService();
     }
 
     @Override
@@ -48,7 +44,8 @@ public void process(Exchange exchange) throws Exception {
         if (XChangeMethod.ticker == method) {
             CurrencyPair pair = exchange.getIn().getHeader(HEADER_CURRENCY_PAIR, CurrencyPair.class);
             pair = pair != null ? pair : exchange.getMessage().getBody(CurrencyPair.class);
-            Ticker ticker = marketService.getTicker(pair);
+            pair = pair != null ? pair : endpoint.getConfiguration().getCurrencyPair();
+            Ticker ticker = endpoint.getTicker(pair);
             exchange.getMessage().setBody(ticker);
         }
     }

components/camel-xchange/src/main/java/org/apache/camel/component/xchange/XChangeMetaDataProducer.java

@@ -55,13 +55,15 @@ else if (XChangeMethod.currencyPairs == method) {
         else if (XChangeMethod.currencyMetaData == method) {
             Currency curr = exchange.getMessage().getHeader(HEADER_CURRENCY, Currency.class);
             curr = curr != null ? curr : exchange.getMessage().getBody(Currency.class);
+            curr = curr != null ? curr : endpoint.getConfiguration().getCurrency();
             Object body = endpoint.getCurrencyMetaData(curr);
             exchange.getMessage().setBody(body);
         } 
 
         else if (XChangeMethod.currencyPairMetaData == method) {
             CurrencyPair pair = exchange.getIn().getHeader(HEADER_CURRENCY_PAIR, CurrencyPair.class);
             pair = pair != null ? pair : exchange.getMessage().getBody(CurrencyPair.class);
+            pair = pair != null ? pair : endpoint.getConfiguration().getCurrencyPair();
             Object body = endpoint.getCurrencyPairMetaData(pair);
             exchange.getMessage().setBody(body);
         }