bitmex authenticate

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexAuthenticator.java

@@ -0,0 +1,38 @@
+package info.bitrich.xchangestream.bitmex;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.nio.charset.Charset;
+
+/**
+ * Created by heath on 2018/3/1.
+ */
+public class BitmexAuthenticator {
+
+    public static String getSHA256String(String str, String key) {
+
+        try {
+            Charset asciiCs = Charset.forName("US-ASCII");
+            SecretKeySpec signingKey = new SecretKeySpec(asciiCs.encode(key).array(), "HmacSHA256");
+            Mac sha256_HMAC = Mac.getInstance("HmacSHA256");
+            sha256_HMAC.init(signingKey);
+            byte[] mac_data = sha256_HMAC.doFinal(asciiCs.encode(str).array());
+            StringBuilder result = new StringBuilder();
+            for (final byte element : mac_data) {
+                result.append(Integer.toString((element & 0xff) + 0x100, 16).substring(1));
+            }
+            // System.out.println("SHA256String Result:[" + result + "]");
+            return result.toString().toUpperCase();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+    public static String generateSignature(String secret, String verb, String url, String nonce, String data) {
+        String message = verb + url + nonce + data;
+        // System.out.println(message);
+        return getSHA256String(message, secret);
+    }
+
+}

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingExchange.java

@@ -24,6 +24,7 @@ public BitmexStreamingExchange() {
     @Override
     protected void initServices() {
         streamingMarketDataService = new BitmexStreamingMarketDataService(streamingService);
+        streamingService.setExchangeSpecification(this.getExchangeSpecification());
     }
 
     @Override

xchange-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingService.java

@@ -1,9 +1,12 @@
 package info.bitrich.xchangestream.bitmex;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import io.reactivex.Completable;
+import io.reactivex.CompletableSource;
+import org.knowm.xchange.ExchangeSpecification;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -23,12 +26,52 @@ public class BitmexStreamingService extends JsonNettyStreamingService {
     private static final Logger LOG = LoggerFactory.getLogger(BitmexStreamingService.class);
     private final ObjectMapper mapper = new ObjectMapper();
 
+    protected ExchangeSpecification exchangeSpecification;
+
     public BitmexStreamingService(String apiUrl) {
         super(apiUrl, Integer.MAX_VALUE);
         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
     }
 
-	 @Override
+    public void setExchangeSpecification(ExchangeSpecification exchangeSpecification) {
+        this.exchangeSpecification = exchangeSpecification;
+    }
+
+    private void login() throws JsonProcessingException {
+        long expires = System.currentTimeMillis() + 30;
+        String apiKey = this.exchangeSpecification.getApiKey();
+        String apiSecret = this.exchangeSpecification.getSecretKey();
+        String path = "/realtime";
+        String signature = BitmexAuthenticator.generateSignature(apiSecret,
+                "GET", path, String.valueOf(expires), "");
+
+        List<Object> args = Arrays.asList(apiKey, expires, signature);
+
+        Map<String, Object> cmd = new HashMap<>();
+        cmd.put("op", "authKey");
+        cmd.put("args", args);
+        this.sendMessage(mapper.writeValueAsString(cmd));
+    }
+
+    @Override
+    public Completable connect() {
+        // Note that we must override connect method in streaming service instead of streaming exchange, because of the auto reconnect feature of NettyStreamingService.
+        // We must ensure the authentication message is also resend when the connection is rebuilt.
+        Completable conn = super.connect();
+        if (this.exchangeSpecification.getApiKey() == null) {
+            return conn;
+        }
+        return conn.andThen((CompletableSource)(completable) -> {
+            try {
+                login();
+                completable.onComplete();
+            } catch (IOException e) {
+                completable.onError(e);
+            }
+        });
+    }
+
+    @Override
     protected void handleMessage(JsonNode message) {
         if (message.has("info") || message.has("success")) {
             return;