OkCoin Websocket Streaming Initial Commit

xchange-examples/src/main/java/com/xeiam/xchange/examples/okcoin/streaming/OkCoinSocketIODemo.java

@@ -0,0 +1,74 @@
+package com.xeiam.xchange.examples.okcoin.streaming;
+
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import com.xeiam.xchange.Exchange;
+import com.xeiam.xchange.ExchangeFactory;
+import com.xeiam.xchange.ExchangeSpecification;
+import com.xeiam.xchange.dto.marketdata.Ticker;
+import com.xeiam.xchange.okcoin.OkCoinExchange;
+import com.xeiam.xchange.okcoin.service.streaming.OkCoinExchangeStreamingConfiguration;
+import com.xeiam.xchange.service.streaming.ExchangeEvent;
+import com.xeiam.xchange.service.streaming.ExchangeEventType;
+import com.xeiam.xchange.service.streaming.StreamingExchangeService;
+
+
+public class OkCoinSocketIODemo {
+
+ public static void main(String[] args) throws InterruptedException {
+
+ ExchangeSpecification exSpec = new ExchangeSpecification(OkCoinExchange.class);
+ exSpec.setSecretKey("aa");
+ exSpec.setApiKey("bb");
+
+ Exchange exchange = ExchangeFactory.INSTANCE.createExchange(exSpec);
+ final StreamingExchangeService service = exchange.getStreamingExchangeService(new OkCoinExchangeStreamingConfiguration());
+
+ Thread consumer = new Thread("consumer") {
+
+ @Override
+ public void run() {
+
+ while (!isInterrupted()) {
+ try {
+ ExchangeEvent event = service.getNextEvent();
+
+ if(event != null) {
+ System.out.println("---> " + event.getPayload() + " " + event.getEventType());
+
+ if(event.getEventType().equals(ExchangeEventType.TICKER)) {
+ Ticker ticker = (Ticker) event.getPayload();
+
+ long x = (new Date()).getTime() - ticker.getTimestamp().getTime();
+ System.out.println("Delay " + x);
+ }
+
+ }
+
+ } catch (InterruptedException e) {
+ this.interrupt();
+ }
+ }
+ }
+
+ };
+
+ // Start consumer.
+ consumer.start();
+
+ // Start streaming service.
+ service.connect();
+
+ // Demonstrate for 30 seconds.
+ TimeUnit.SECONDS.sleep(30);
+
+ // Disconnect streaming service.
+ service.disconnect();
+
+ // Interrupt consumer to exit.
+ consumer.interrupt();
+ }
+
+
+}

xchange-okcoin/pom.xml

@@ -1,4 +1,5 @@
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 
  <modelVersion>4.0.0</modelVersion>
 
@@ -29,6 +30,12 @@
  <version>${project.version}</version>
  </dependency>
 
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ <version>4.1.0.Beta3</version>
+ </dependency>
+
  </dependencies>
 
 </project>

xchange-okcoin/src/main/java/com/xeiam/xchange/okcoin/OkCoinExchange.java

@@ -10,9 +10,12 @@
 import com.xeiam.xchange.okcoin.service.polling.OkCoinFuturesTradeService;
 import com.xeiam.xchange.okcoin.service.polling.OkCoinMarketDataService;
 import com.xeiam.xchange.okcoin.service.polling.OkCoinTradeService;
+import com.xeiam.xchange.okcoin.service.streaming.OkCoinStreamingExchangeService;
+import com.xeiam.xchange.service.streaming.ExchangeStreamingConfiguration;
+import com.xeiam.xchange.service.streaming.StreamingExchangeService;
 
 public class OkCoinExchange extends BaseExchange {
-
+ 
  @Override
  public void applySpecification(ExchangeSpecification exchangeSpecification) {
 
@@ -43,6 +46,7 @@ public void applySpecification(ExchangeSpecification exchangeSpecification) {
  if (exchangeSpecification.getExchangeSpecificParametersItem("Use_Intl").equals(true)) {
  exchangeSpecification.setSslUri("https://www.okcoin.com/api");
  exchangeSpecification.setHost("www.okcoin.com");
+ exchangeSpecification.setExchangeSpecificParametersItem("Websocket_SslUri", "wss://real.okcoin.com:10440/websocket/okcoinapi");
  }
 
  }
@@ -60,9 +64,16 @@ public ExchangeSpecification getDefaultExchangeSpecification() {
  exchangeSpecification.setExchangeSpecificParametersItem("Use_Intl", false);
  exchangeSpecification.setExchangeSpecificParametersItem("Use_Futures", false);
 
+ exchangeSpecification.setExchangeSpecificParametersItem("Websocket_SslUri", "wss://real.okcoin.cn:10440/websocket/okcoinapi");
+
  return exchangeSpecification;
  }
 
+ @Override
+ public StreamingExchangeService getStreamingExchangeService(ExchangeStreamingConfiguration configuration) {
+ return new OkCoinStreamingExchangeService(getExchangeSpecification(), configuration);
+ }
+
  @Override
  public SynchronizedValueFactory<Long> getNonceFactory() {
  // This exchange doesn't use a nonce for authentication

xchange-okcoin/src/main/java/com/xeiam/xchange/okcoin/dto/marketdata/OkCoinDepth.java

@@ -10,7 +10,7 @@ public class OkCoinDepth {
  private final BigDecimal[][] asks;
  private final BigDecimal[][] bids;
 
- public OkCoinDepth(@JsonProperty("asks") final BigDecimal[][] asks, @JsonProperty("bids") final BigDecimal[][] bids) {
+ public OkCoinDepth(@JsonProperty("asks") final BigDecimal[][] asks, @JsonProperty("bids") final BigDecimal[][] bids, @JsonProperty(required=false, value="timestamp") long timestamp) {
 
  this.asks = asks;
  this.bids = bids;

xchange-okcoin/src/main/java/com/xeiam/xchange/okcoin/dto/marketdata/OkCoinStreamingTicker.java

@@ -0,0 +1,84 @@
+package com.xeiam.xchange.okcoin.dto.marketdata;
+
+import java.math.BigDecimal;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class OkCoinStreamingTicker {
+
+ private final BigDecimal high;
+ private final BigDecimal low;
+ private final BigDecimal buy;
+ private final BigDecimal sell;
+ private final BigDecimal last;
+ private final String vol;
+ private final long timestamp;
+
+ public OkCoinStreamingTicker(@JsonProperty("high") final BigDecimal high, @JsonProperty("low") final BigDecimal low, @JsonProperty("buy") final BigDecimal buy, @JsonProperty("sell") final BigDecimal sell,
+ @JsonProperty("last") final BigDecimal last, @JsonProperty("vol") final String vol, @JsonProperty("timestamp") long timestamp) {
+
+ this.high = high;
+ this.low = low;
+ this.buy = buy;
+ this.sell = sell;
+ this.last = last;
+ this.vol = vol;
+ this.timestamp = timestamp;
+ }
+
+ /**
+ * @return the high
+ */
+ public BigDecimal getHigh() {
+
+ return high;
+ }
+
+ /**
+ * @return the low
+ */
+ public BigDecimal getLow() {
+
+ return low;
+ }
+
+ /**
+ * @return the buy
+ */
+ public BigDecimal getBuy() {
+
+ return buy;
+ }
+
+ /**
+ * @return the sell
+ */
+ public BigDecimal getSell() {
+
+ return sell;
+ }
+
+ /**
+ * @return the last
+ */
+ public BigDecimal getLast() {
+
+ return last;
+ }
+
+ /**
+ * @return the vol
+ */
+ public String getVol() {
+
+ return vol;
+ }
+
+ /**
+ * @return the timestamp
+ */ 
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+}

xchange-okcoin/src/main/java/com/xeiam/xchange/okcoin/service/streaming/MD5Util.java

@@ -0,0 +1,76 @@
+package com.xeiam.xchange.okcoin.service.streaming;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+public class MD5Util {
+
+ public static String buildMysignV1(Map<String, String> sArray,
+ String secretKey) {
+ String mysign = "";
+ try {
+ String prestr = createLinkString(sArray); 
+ prestr = prestr + "&secret_key=" + secretKey; 
+ System.out.println("prestr "+prestr);
+ mysign = getMD5String(prestr);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return mysign;
+ }
+
+ public static String createLinkString(Map<String, String> params) {
+
+ List<String> keys = new ArrayList<String>(params.keySet());
+ Collections.sort(keys);
+ String prestr = "";
+ for (int i = 0; i < keys.size(); i++) {
+ String key = keys.get(i);
+ String value = params.get(key);
+ if (i == keys.size() - 1) {
+ prestr = prestr + key + "=" + value;
+ } else {
+ prestr = prestr + key + "=" + value + "&";
+ }
+ }
+ return prestr;
+ }
+
+ private static final char HEX_DIGITS[] = { '0', '1', '2', '3', '4', '5',
+ '6', '7', '8', '9', 'A', 'B', 'C', 'D', 'E', 'F' };
+
+ public static String getMD5String(String str) {
+ try {
+ if (str == null || str.trim().length() == 0) {
+ return "";
+ }
+ byte[] bytes = str.getBytes();
+ MessageDigest messageDigest = MessageDigest.getInstance("MD5");
+ messageDigest.update(bytes);
+ bytes = messageDigest.digest();
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i < bytes.length; i++) {
+ sb.append(HEX_DIGITS[(bytes[i] & 0xf0) >> 4] + ""
+ + HEX_DIGITS[bytes[i] & 0xf]);
+ }
+ return sb.toString();
+ } catch (NoSuchAlgorithmException e) {
+ e.printStackTrace();
+ }
+ return "";
+ }
+
+ public static String getParams(Map<String,String> map){
+ StringBuilder params = new StringBuilder("{");
+ for(Entry<String,String> param:map.entrySet()){
+ params.append("'").append(param.getKey()).append("':'").append(param.getValue()).append("',");
+ }
+ params.replace(params.length()-1,params.length(),"}");
+ return params.toString();
+ }
+}

xchange-okcoin/src/main/java/com/xeiam/xchange/okcoin/service/streaming/OkCoinExchangeEvent.java

@@ -0,0 +1,31 @@
+package com.xeiam.xchange.okcoin.service.streaming;
+
+import com.xeiam.xchange.service.streaming.ExchangeEvent;
+import com.xeiam.xchange.service.streaming.ExchangeEventType;
+
+class OkCoinExchangeEvent implements ExchangeEvent {
+ private final String rawData;
+ private final Object payload;
+ private final ExchangeEventType exchangeEventType;
+
+ public OkCoinExchangeEvent(ExchangeEventType exchangeEventType, Object payload) {
+ this.rawData = null;
+ this.payload = payload;
+ this.exchangeEventType = exchangeEventType;
+ }
+
+ @Override
+ public Object getPayload() {
+ return payload;
+ }
+
+ @Override
+ public String getData() {
+ return rawData;
+ }
+
+ @Override
+ public ExchangeEventType getEventType() {
+ return exchangeEventType;
+ }
+}

xchange-okcoin/src/main/java/com/xeiam/xchange/okcoin/service/streaming/OkCoinExchangeStreamingConfiguration.java

@@ -0,0 +1,47 @@
+package com.xeiam.xchange.okcoin.service.streaming;
+
+import com.xeiam.xchange.currency.CurrencyPair;
+import com.xeiam.xchange.service.streaming.ExchangeStreamingConfiguration;
+
+
+public class OkCoinExchangeStreamingConfiguration implements ExchangeStreamingConfiguration {
+
+ private final CurrencyPair[] marketDataCurrencyPairs;
+
+ public OkCoinExchangeStreamingConfiguration() {
+ marketDataCurrencyPairs = new CurrencyPair[]{ CurrencyPair.BTC_CNY };
+ }
+
+ public OkCoinExchangeStreamingConfiguration(CurrencyPair[] marketDataCurrencyPairs) {
+ this.marketDataCurrencyPairs = marketDataCurrencyPairs;
+ } 
+
+ @Override
+ public int getMaxReconnectAttempts() {
+ return 0;
+ }
+
+ @Override
+ public int getReconnectWaitTimeInMs() {
+ return 0;
+ }
+
+ @Override
+ public int getTimeoutInMs() {
+ return 0;
+ }
+
+ @Override
+ public boolean isEncryptedChannel() {
+ return false;
+ }
+
+ @Override
+ public boolean keepAlive() {
+ return false;
+ }
+
+ public CurrencyPair[] getMarketDataCurrencyPairs() {
+ return marketDataCurrencyPairs;
+ }
+}