changes to add reconnection to ticker and added new pojo for indices quote

sujith · sujith · commit 4da9e55350b9 · 2017-04-11T17:08:04.000+05:30
diff --git a/src/main/java/com/rainmatter/kiteconnect/KiteConnect.java b/src/main/java/com/rainmatter/kiteconnect/KiteConnect.java
@@ -421,12 +421,12 @@ public Quote getQuote(String exchange, String tradingSymbol) throws KiteExceptio
      *
      * @param exchange  Exchange in which instrument is listed. exchange can be NSE, BSE.
      * @param tradingSymbol Tradingsymbol of the instrument (ex. NIFTY 50).
-     * @return Quote object.
+     * @return IndicesQuote object.
      */
-    public Quote getQuoteIndices(String exchange, String tradingSymbol) throws KiteException, JSONException {
+    public IndicesQuote getQuoteIndices(String exchange, String tradingSymbol) throws KiteException, JSONException {
         Map<String, Object> params = new HashMap<String, Object>();
         KiteRequest kiteRequest = new KiteRequest();
-        return new Quote().parseIndicesResponse(kiteRequest.getRequest(routes.get("market.quote").replace(":exchange", exchange).replace(":tradingsymbol", tradingSymbol), authorize(params)));
+        return new IndicesQuote().parseIndicesResponse(kiteRequest.getRequest(routes.get("market.quote").replace(":exchange", exchange).replace(":tradingsymbol", tradingSymbol), authorize(params)));
     }
 
     /**
diff --git a/src/main/java/com/rainmatter/models/IndicesQuote.java b/src/main/java/com/rainmatter/models/IndicesQuote.java
@@ -0,0 +1,37 @@
+package com.rainmatter.models;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.annotations.SerializedName;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+/**
+ * Created by sujith on 4/11/17.
+ */
+public class IndicesQuote {
+
+    @SerializedName("change")
+    public double change;
+    @SerializedName("change_percent")
+    public double changePercent;
+    @SerializedName("last_price")
+    public double lastPrice;
+    @SerializedName("close")
+    public double close;
+    @SerializedName("open")
+    public double open;
+
+    /**
+     * Parses response from quotes api.
+     * @return Quote is POJO sent back to user.
+     * @param response json response from server.
+     * @throws JSONException json parsing error.
+     */
+    public IndicesQuote parseIndicesResponse(JSONObject response) throws JSONException {
+        GsonBuilder gsonBuilder = new GsonBuilder();
+        Gson gson = gsonBuilder.create();
+        IndicesQuote indicesQuote = gson.fromJson(String.valueOf(response.get("data")), IndicesQuote.class);
+        return indicesQuote;
+    }
+}
diff --git a/src/main/java/com/rainmatter/ticker/KiteTicker.java b/src/main/java/com/rainmatter/ticker/KiteTicker.java
@@ -1,10 +1,15 @@
 package com.rainmatter.ticker;
 
+/**
+ * Created by H1ccup on 10/09/16.
+ */
+
 import com.neovisionaries.ws.client.*;
-import com.rainmatter.models.Depth;
-import com.rainmatter.models.Tick;
 import com.rainmatter.kiteconnect.KiteConnect;
 import com.rainmatter.kiteconnect.Routes;
+import com.rainmatter.kitehttp.exceptions.KiteException;
+import com.rainmatter.models.Depth;
+import com.rainmatter.models.Tick;
 import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;
@@ -15,7 +20,8 @@
 import java.util.*;
 
 /**
- * Provides capability to establish a websocket connection to get live quotes.
+ * Ticker provider sends tokens to com.rainmatter.ticker server and get ticks from com.rainmatter.ticker server. Ticker server sends data in bytes. This Class
+ * gets ticks and converts into readable format which includes our own business logic.
  */
 public class KiteTicker {
 
@@ -25,8 +31,8 @@ public class KiteTicker {
     private OnDisconnect onDisconnectedListener;
     private WebSocket ws;
 
-    //private Thread mThread;
     private KiteConnect _kiteSdk;
+    private KiteTicker tickerProvider;
 
     public final int NseCM = 1,
             NseFO = 2,
@@ -42,12 +48,60 @@ public class KiteTicker {
             mUnSubscribe = "unsubscribe",
             mSetMode = "mode";
 
-
-    private boolean connection = false;
     public static String modeFull  = "full", // Full quote inluding market depth. 172 bytes.
             modeQuote = "quote", // Quote excluding market depth. 52 bytes.
             modeLTP   = "ltp"; // Only LTP. 4 bytes.;
 
+    private long lastTickArrivedAt = 0;
+    private long timeIntervalForReconnection = 5;
+    private Set<Long> subscribedTokens = new HashSet<>();
+    private int maxRetries = 50;
+    private int count = 0;
+    private Timer timer = null;
+    private boolean tryReconnection = false;
+    
+    /** Returns task which performs check every second for reconnection*/
+    private TimerTask getTask(){
+        TimerTask checkForRestartTask = new TimerTask() {
+            @Override
+            public void run() {
+                Date currentDate = new Date();
+                long timeInterval = (currentDate.getTime() - lastTickArrivedAt)/1000;
+                if(lastTickArrivedAt > 0) {
+                    if (timeInterval >= timeIntervalForReconnection) {
+                        if(maxRetries == -1 || count <= maxRetries) {
+                            reconnect(new ArrayList<>(subscribedTokens));
+                            count++;
+                        }else if(maxRetries != -1 && count > maxRetries){
+                            if(timer != null){
+                                timer.cancel();
+                            }
+                        }
+                    }
+                }
+            }
+        };
+        return checkForRestartTask;
+    }
+
+    /** Set tryReconnection, to instruct KiteTicker that it has to reconnect, if ticker is disconnected*/
+    public void setTryReconnection(boolean retry){
+        tryReconnection = retry;
+    }
+
+    /** Set minimum time interval after which ticker has to restart in seconds*/
+    public void setTimeIntervalForReconnection(int interval) throws KiteException {
+        if(interval >= 5) {
+            timeIntervalForReconnection = interval;
+        }else
+            throw new KiteException("reconnection interval can't be less than five seconds", 00);
+    }
+
+    /** Set max number of retries for reconnection, for infinite retries set value as -1 */
+    public void setMaxRetries(int maxRetries){
+        this.maxRetries = maxRetries;
+    }
+
     public KiteTicker(KiteConnect kiteSdk){
         _kiteSdk = kiteSdk;
         createUrl();
@@ -86,9 +140,8 @@ public void setOnDisconnectedListener(OnDisconnect listener){
     /** Establishes a web socket connection.*/
     public void connect() throws WebSocketException, IOException {
 
-        if(connection){
-            disconnect();
-            connection = false;
+        if(isConnectionOpen()){
+            return;
         }
 
         if(wsuri == null){
@@ -99,10 +152,18 @@ public void connect() throws WebSocketException, IOException {
 
             @Override
             public void onConnected(WebSocket websocket, Map<String, List<String>> headers) {
-                connection = true;
+                //connection = true;
                 if(onConnectedListener != null) {
                     onConnectedListener.onConnected();
                 }
+                if(tryReconnection) {
+                    if (timer != null) {
+                        timer.cancel();
+                    }
+                    timer = new Timer();
+                    timer.scheduleAtFixedRate(getTask(), 0, 1000);
+
+                }
             }
 
             @Override
@@ -118,6 +179,9 @@ public void onBinaryMessage(WebSocket websocket, byte[] binary)  {
                  if (onTickerArrivalListener != null) {
                       onTickerArrivalListener.onTick(tickerData);
                  }
+
+                 Date date = new Date();
+                 lastTickArrivedAt = date.getTime();
             }
 
             /**
@@ -133,24 +197,35 @@ public void onDisconnected(WebSocket websocket, WebSocketFrame serverCloseFrame,
                 if(onDisconnectedListener != null){
                     onDisconnectedListener.onDisconnected();
                 }
-                    return;
+                //connection = false;
+                return;
              }
 
              @Override
              public void onError(WebSocket websocket, WebSocketException cause) {
                     super.onError(websocket, cause);
-                }
+             }
+
         });
             ws.connect();
     }
 
     /** Disconnects websocket connection.*/
     public void disconnect(){
+        if(timer != null){
+            timer.cancel();
+        }
         if (ws != null && ws.isOpen()) {
             ws.disconnect();
+            subscribedTokens = new HashSet<>();
         }
     }
 
+    /** Disconnects websocket connection only for internal use*/
+    private void nonUserDisconnect(){
+       ws.disconnect();
+    }
+
     /** Returns true if websocket connection is open.
      * @return boolean*/
     public boolean isConnectionOpen(){
@@ -192,16 +267,17 @@ public void setMode(ArrayList<Long> tokens, String mode){
 
     /** Subscribes for list of tokens.
      * @param tokens is list of tokens to be subscribed for.*/
-    public void subscribe(ArrayList<Long> tokens) throws IOException, WebSocketException {
+    public void subscribe(ArrayList<Long> tokens) throws IOException, WebSocketException, KiteException {
         if(ws != null) {
             if (ws.isOpen()) {
                 createTickerJsonObject(tokens, mSubscribe);
                 ws.sendText(createTickerJsonObject(tokens, mSubscribe).toString());
                 setMode(tokens, modeQuote);
             }else
-                connect();
+                throw new KiteException("ticker is not connected", 504);
         }else
-            connect();
+            throw new KiteException("ticker is not connected", 504);
+        subscribedTokens.addAll(tokens);
     }
 
     private JSONObject createTickerJsonObject(ArrayList<Long> tokens, String action) {
@@ -225,6 +301,7 @@ public void unsubscribe(ArrayList<Long> tokens){
         if(ws != null) {
             if (ws.isOpen()) {
                 ws.sendText(createTickerJsonObject(tokens, mUnSubscribe).toString());
+                subscribedTokens.removeAll(tokens);
             }
         }
     }
@@ -299,14 +376,13 @@ private Tick getNseIndeciesData(byte[] bin, int x){
             tick.setOpenPrice(convertToDouble(getBytes(bin, 16, 20)) / dec);
             tick.setClosePrice(convertToDouble(getBytes(bin, 20, 24)) / dec);
             tick.setNetPriceChangeFromClosingPrice(convertToDouble(getBytes(bin, 24, 28)) / dec);
-            return tick;
         }else {
             tick.setMode(modeLTP);
             tick.setTradable(false);
             tick.setToken(x);
             tick.setLastTradedPrice(convertToDouble(getBytes(bin, 4, 8)) / dec);
-            return tick;
         }
+        return tick;
     }
 
     private Tick getLtpQuote(byte[] bin, int x, int dec1){
@@ -412,4 +488,33 @@ private int getLengthFromByteArray(byte[] bin){
         return bb.getShort();
     }
 
+    /** Disconnects and reconnects ticker*/
+    private void reconnect(final ArrayList<Long> tokens) {
+        try {
+            nonUserDisconnect();
+            connect();
+            lastTickArrivedAt = 0;
+            final OnConnect onUsersConnectedListener = this.onConnectedListener;
+            setOnConnectedListener(new OnConnect() {
+                @Override
+                public void onConnected() {
+                    try {
+                        subscribe(tokens);
+                        onConnectedListener = onUsersConnectedListener;
+                    } catch (IOException e) {
+                        e.printStackTrace();
+                    } catch (WebSocketException e) {
+                        e.printStackTrace();
+                    } catch (KiteException e) {
+                        e.printStackTrace();
+                    }
+                }
+            });
+        }catch (WebSocketException we){
+            we.printStackTrace();
+        }catch (IOException ie){
+            ie.printStackTrace();
+        }
+    }
+
 }
