From 5c918d834bff568a8551026a6ec5b58fb2136ab7 Mon Sep 17 00:00:00 2001 From: Sachin Date: Wed, 5 Dec 2018 03:06:57 +0530 Subject: [PATCH] Removed old websocket client library (nv-websocket-client), added OKHttpWebsocket library as a new websocket library. Updated Socket, BasicListener and reconnectionstrategy to support new version of the library. --- build.gradle | 2 +- src/main/java/Main.java | 157 ++--- .../java/io/github/sac/BasicListener.java | 13 +- src/main/java/io/github/sac/Constants.java | 6 + .../java/io/github/sac/ReconnectStrategy.java | 101 ---- .../io/github/sac/ReconnectionStrategy.java | 42 ++ src/main/java/io/github/sac/Socket.java | 564 ++++++++++-------- src/main/java/io/github/sac/State.java | 9 + src/main/java/io/github/sac/TaskHandler.java | 57 ++ 9 files changed, 510 insertions(+), 441 deletions(-) create mode 100644 src/main/java/io/github/sac/Constants.java delete mode 100644 src/main/java/io/github/sac/ReconnectStrategy.java create mode 100644 src/main/java/io/github/sac/ReconnectionStrategy.java create mode 100644 src/main/java/io/github/sac/State.java create mode 100644 src/main/java/io/github/sac/TaskHandler.java diff --git a/build.gradle b/build.gradle index 2d3e211..2f3e074 100644 --- a/build.gradle +++ b/build.gradle @@ -80,6 +80,6 @@ bintray{ } dependencies { - compile 'com.neovisionaries:nv-websocket-client:1.30' + compile 'com.squareup.okhttp3:okhttp:3.9.0' compile group: 'org.json', name: 'json', version: '20090211' } diff --git a/src/main/java/Main.java b/src/main/java/Main.java index fd83f4e..c7f545a 100644 --- a/src/main/java/Main.java +++ b/src/main/java/Main.java @@ -1,9 +1,6 @@ -import com.neovisionaries.ws.client.WebSocketException; -import com.neovisionaries.ws.client.WebSocketFrame; import io.github.sac.*; - -import java.util.List; -import java.util.Map; +import okhttp3.Headers; +import okhttp3.Response; /** * Created by sachin on 8/11/16. @@ -19,24 +16,91 @@ public static void main(String arg[]) { socket.setListener(new BasicListener() { - public void onConnected(Socket socket,Map> headers) { - System.out.println("Connected to endpoint"); + + public void onSetAuthToken(String token, Socket socket) { + System.out.println("Set auth token got called"); + socket.setAuthToken(token); } - public void onDisconnected(Socket socket,WebSocketFrame serverCloseFrame, WebSocketFrame clientCloseFrame, boolean closedByServer) { - System.out.println("Disconnected from end-point"); + @Override + public void onConnected(Socket socket, Headers headers) { + System.out.println("Connected to endpoint"); + + + socket.emit("chat","Hi"); + socket.emit("chat", "Hi", new Ack() { + @Override + public void call(String eventName, Object error, Object data) { + System.out.println("Got message for :"+eventName+" error is :"+error+" data is :"+data); + } + }); + + socket.on("yell", new Emitter.Listener() { + @Override + public void call(String eventName, Object data) { + System.out.println("Got message for :"+eventName+" data is :"+data); + } + }); + + socket.on("yell", new Emitter.AckListener() { + @Override + public void call(String eventName, Object data, Ack ack) { + System.out.println("Got message for :"+eventName+" data is :"+data); + //sending ack back + + ack.call(eventName,"This is error","This is data"); + } + }); +// +// + Socket.Channel channel = socket.createChannel("yell"); +// + channel.subscribe(new Ack() { + @Override + public void call(String channelName, Object error, Object data) { + if (error==null){ + System.out.println("Subscribed to channel "+channelName+" successfully"); + } + } + }); + + channel.publish("Hi sachin", new Ack() { + @Override + public void call(String channelName, Object error, Object data) { + if (error==null){ + System.out.println("Published message to channel "+channelName+" successfully"); + } + } + }); + + channel.onMessage(new Emitter.Listener() { + @Override + public void call(String channelName, Object data) { + + System.out.println("Got message for channel "+channelName+" data is "+data); + } + }); + + channel.unsubscribe(new Ack() { + @Override + public void call(String name, Object error, Object data) { + System.out.println("Unsubscribed successfully for channel " + name); + } + }); +// channel.unsubscribe(); } - public void onConnectError(Socket socket,WebSocketException exception) { - System.out.println("Got connect error "+ exception); + @Override + public void onDisconnected(Socket socket, int code, String reason) { + System.out.println("Disconnected from end-point"); } - public void onSetAuthToken(String token, Socket socket) { - System.out.println("Set auth token got called"); - socket.setAuthToken(token); + @Override + public void onConnectError(Socket socket, Throwable throwable, Response response) { + System.out.println("Got connect error "+ throwable.getMessage()); } - public void onAuthentication(Socket socket,Boolean status) { + public void onAuthentication(Socket socket, Boolean status) { if (status) { System.out.println("socket is authenticated"); } else { @@ -46,7 +110,7 @@ public void onAuthentication(Socket socket,Boolean status) { }); - socket.setReconnection(new ReconnectStrategy().setDelay(3000).setMaxAttempts(10)); //Connect after each 2 seconds for 30 times + socket.setReconnection(new ReconnectionStrategy(10, 3000)); //Connect after each 2 seconds for 30 times socket.connectAsync(); @@ -55,67 +119,6 @@ public void onAuthentication(Socket socket,Boolean status) { socket.disableLogging(); - socket.emit("chat","Hi"); - socket.emit("chat", "Hi", new Ack() { - @Override - public void call(String eventName, Object error, Object data) { - System.out.println("Got message for :"+eventName+" error is :"+error+" data is :"+data); - } - }); - - socket.on("yell", new Emitter.Listener() { - @Override - public void call(String eventName, Object data) { - System.out.println("Got message for :"+eventName+" data is :"+data); - } - }); - - socket.on("yell", new Emitter.AckListener() { - @Override - public void call(String eventName, Object data, Ack ack) { - System.out.println("Got message for :"+eventName+" data is :"+data); - //sending ack back - - ack.call(eventName,"This is error","This is data"); - } - }); -// -// - Socket.Channel channel = socket.createChannel("yell"); -// - channel.subscribe(new Ack() { - @Override - public void call(String channelName, Object error, Object data) { - if (error==null){ - System.out.println("Subscribed to channel "+channelName+" successfully"); - } - } - }); - - channel.publish("Hi sachin", new Ack() { - @Override - public void call(String channelName, Object error, Object data) { - if (error==null){ - System.out.println("Published message to channel "+channelName+" successfully"); - } - } - }); - - channel.onMessage(new Emitter.Listener() { - @Override - public void call(String channelName, Object data) { - - System.out.println("Got message for channel "+channelName+" data is "+data); - } - }); - - channel.unsubscribe(new Ack() { - @Override - public void call(String name, Object error, Object data) { - System.out.println("Unsubscribed successfully"); - } - }); - channel.unsubscribe(); // channel.subscribe(new Ack() { // @Override diff --git a/src/main/java/io/github/sac/BasicListener.java b/src/main/java/io/github/sac/BasicListener.java index 10c35d8..b410348 100644 --- a/src/main/java/io/github/sac/BasicListener.java +++ b/src/main/java/io/github/sac/BasicListener.java @@ -1,19 +1,18 @@ package io.github.sac; -import com.neovisionaries.ws.client.WebSocketException; -import com.neovisionaries.ws.client.WebSocketFrame; -import java.util.List; -import java.util.Map; +import okhttp3.Headers; +import okhttp3.Response; /** * Created by sachin on 13/11/16. */ public interface BasicListener { - void onConnected(Socket socket, Map> headers); - void onDisconnected(Socket socket, WebSocketFrame serverCloseFrame, WebSocketFrame clientCloseFrame, boolean closedByServer); + void onConnected(Socket socket, Headers headers); - void onConnectError(Socket socket, WebSocketException exception); + void onDisconnected(Socket socket, int code, String reason); + + void onConnectError(Socket socket, Throwable throwable, Response response); void onAuthentication(Socket socket, Boolean status); diff --git a/src/main/java/io/github/sac/Constants.java b/src/main/java/io/github/sac/Constants.java new file mode 100644 index 0000000..6e1199f --- /dev/null +++ b/src/main/java/io/github/sac/Constants.java @@ -0,0 +1,6 @@ +package io.github.sac; + +public class Constants { + public static final String PING_MESSAGE = "#1"; + public static final String PONG_MESSAGE = "#2"; +} diff --git a/src/main/java/io/github/sac/ReconnectStrategy.java b/src/main/java/io/github/sac/ReconnectStrategy.java deleted file mode 100644 index 3cf31ff..0000000 --- a/src/main/java/io/github/sac/ReconnectStrategy.java +++ /dev/null @@ -1,101 +0,0 @@ -package io.github.sac; - -import java.util.logging.Level; -import java.util.logging.Logger; - -/** - * Created by sachin on 16/11/16. - */ - -public class ReconnectStrategy { - - private final static Logger LOGGER = Logger.getLogger(ReconnectStrategy.class.getName()); - /** - * The number of milliseconds to delay before attempting to reconnect. - * Default: 2000 - */ - - int reconnectInterval; - - /** - * The maximum number of milliseconds to delay a reconnection attempt. - * Default: 30000 - */ - - int maxReconnectInterval; - - /** - * The rate of increase of the reconnect delay. Allows reconnect attempts to back off when problems persist. - * Default: 1 - */ - - float reconnectDecay; - - /** - * The maximum number of reconnection attempts that will be made before giving up. If null, reconnection attempts - * will be continue to be made forever. - * Default: null - */ - - Integer maxAttempts; - - Integer attemptsMade; - - - public ReconnectStrategy() { - LOGGER.setLevel(Level.INFO); - reconnectInterval = 2000; - maxReconnectInterval = 30000; - reconnectDecay = (float) 1; - maxAttempts = null; //forever - attemptsMade = 0; - } - - public ReconnectStrategy setMaxAttempts(Integer maxAttempts) { - this.maxAttempts = maxAttempts; - return this; - } - - public ReconnectStrategy setDelay(int delay) { - reconnectInterval = delay; - return this; - } - - public void setAttemptsMade(Integer attemptsMade) { - this.attemptsMade = attemptsMade; - } - - public ReconnectStrategy(int reconnectInterval, int maxReconnectInterval, float reconnectDecay, int maxAttempts) { - if (reconnectInterval > maxReconnectInterval) { - this.reconnectInterval = maxReconnectInterval; - } else { - this.reconnectInterval = reconnectInterval; - } - this.maxReconnectInterval = maxReconnectInterval; - this.reconnectDecay = reconnectDecay; - this.maxAttempts = maxAttempts; - attemptsMade = 0; - } - - - public void processValues() { - attemptsMade++; - LOGGER.info("Attempt number :" + attemptsMade); - if (reconnectInterval < maxReconnectInterval) { - reconnectInterval = (int) (reconnectInterval * reconnectDecay); - if (reconnectInterval > maxReconnectInterval) { - reconnectInterval = maxReconnectInterval; - } - } - } - - public int getReconnectInterval() { - return reconnectInterval; - } - - - public boolean areAttemptsComplete() { - return attemptsMade.equals(maxAttempts); - } - -} diff --git a/src/main/java/io/github/sac/ReconnectionStrategy.java b/src/main/java/io/github/sac/ReconnectionStrategy.java new file mode 100644 index 0000000..c69298a --- /dev/null +++ b/src/main/java/io/github/sac/ReconnectionStrategy.java @@ -0,0 +1,42 @@ +package io.github.sac; + +/** + * Created by sachin on 15/6/17. + */ + +public class ReconnectionStrategy { + private int MaxAttempts; + private int numberOfAttempts; + private int reconnectInterval; + private int maxReconnectInterval = 30000; + + public ReconnectionStrategy(int maxAttempts, int reconnectInterval) { + MaxAttempts = maxAttempts; + if (reconnectInterval < maxReconnectInterval) { + this.reconnectInterval = reconnectInterval; + } else { + this.reconnectInterval = maxReconnectInterval; + } + numberOfAttempts = 0; + } + + public int getNumberOfAttempts() { + return numberOfAttempts; + } + + public void setNumberOfAttempts(int numberOfAttempts) { + this.numberOfAttempts = numberOfAttempts; + } + + public int getReconnectInterval() { + return reconnectInterval; + } + + public void processAttempts() { + numberOfAttempts++; + } + + public int getMaxAttempts() { + return MaxAttempts; + } +} diff --git a/src/main/java/io/github/sac/Socket.java b/src/main/java/io/github/sac/Socket.java index 5940501..a10fb39 100644 --- a/src/main/java/io/github/sac/Socket.java +++ b/src/main/java/io/github/sac/Socket.java @@ -1,17 +1,11 @@ package io.github.sac; -import com.neovisionaries.ws.client.OpeningHandshakeException; -import com.neovisionaries.ws.client.StatusLine; -import com.neovisionaries.ws.client.WebSocket; -import com.neovisionaries.ws.client.WebSocketAdapter; -import com.neovisionaries.ws.client.WebSocketException; -import com.neovisionaries.ws.client.WebSocketFactory; -import com.neovisionaries.ws.client.WebSocketFrame; -import com.neovisionaries.ws.client.WebSocketState; -import java.io.IOException; -import java.util.ArrayList; +import okhttp3.*; +import okio.ByteString; +import org.json.JSONException; +import org.json.JSONObject; + import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.Timer; import java.util.TimerTask; @@ -19,49 +13,60 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.logging.Level; import java.util.logging.Logger; -import org.json.JSONException; -import org.json.JSONObject; /** - * Created by sachin on 13/11/16. + * Created by sachin on 7/6/17. */ -public class Socket extends Emitter { +public /*final*/ class Socket extends Emitter { private final Logger logger = Logger.getLogger(Socket.class.getName()); private AtomicInteger counter; - private String URL; - private WebSocketFactory factory; - private ReconnectStrategy strategy; - private WebSocket ws; private BasicListener listener; + private OkHttpClient client; + private String url; + private TaskHandler pingHandler; + private TaskHandler timeoutHandler; + private long pingInterval; + private WebSocket ws; + private State currentState = State.DISCONNECTED; + + private ReconnectionStrategy strategy; + private Timer timer; + private boolean selfDisconnect; + private boolean pingEnable; private String AuthToken; private HashMap acks; private ConcurrentHashMap channels; - private WebSocketAdapter adapter; - private Map headers; + private Headers headers; - public Socket(String URL) { - this.URL = URL; - factory = new WebSocketFactory().setConnectionTimeout(5000); + public Socket(String url) { + this.url = url; + this.client = new OkHttpClient(); counter = new AtomicInteger(1); acks = new HashMap<>(); channels = new ConcurrentHashMap<>(); - adapter = getAdapter(); - headers = new HashMap<>(); - putDefaultHeaders(); + + setState(State.DISCONNECTED); + selfDisconnect = false; + pingEnable = false; + pingInterval = 2000; + pingHandler = new TaskHandler(); + timeoutHandler = new TaskHandler(); + headers = getDefaultHeadersBuilder().build(); } - private void putDefaultHeaders() { - headers.put("Accept-Encoding", "gzip, deflate, sdch"); - headers.put("Accept-Language", "en-US,en;q=0.8"); - headers.put("Pragma", "no-cache"); - headers.put("User-Agent", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.87 Safari/537.36"); + private Headers.Builder getDefaultHeadersBuilder() { + return new Headers.Builder() + .add("Accept-Encoding", "gzip, deflate, sdch") + .add("Accept-Language", "en-US,en;q=0.8") + .add("Pragma", "no-cache") + .add("User-Agent", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.87 Safari/537.36"); } - public Channel createChannel(String name) { - if(channels.containsKey(name)){ + public Socket.Channel createChannel(String name) { + if (channels.containsKey(name)) { return channels.get(name); } @@ -75,166 +80,213 @@ public ConcurrentHashMap getChannels() { } public Channel getChannelByName(String name) { - channels.get(name); + if (channels.containsKey(name)) { + return channels.get(name); + } return null; } - public void seturl(String url) { - this.URL = url; + public void setUrl(String url) { + this.url = url; } - public void setReconnection(ReconnectStrategy strategy) { + public void setReconnection(ReconnectionStrategy strategy) { this.strategy = strategy; } + public void setListener(BasicListener listener) { this.listener = listener; } - public Logger getLogger(){ - return logger; - } - /** - * used to set up TLS/SSL connection to server for more details visit neovisionaries websocket client - */ - public WebSocketFactory getFactorySettings() { - return factory; + public Logger getLogger() { + return logger; } public void setAuthToken(String token) { AuthToken = token; } - public WebSocketAdapter getAdapter() { - return new WebSocketAdapter() { - + public WebSocketListener getWebscoketListener() { + return new WebSocketListener() { + // OkHttp WebSocket callbacks @Override - public void onConnected(WebSocket websocket, Map> headers) throws Exception { - + public void onOpen(WebSocket webSocket, Response response) { + logger.info("Connected to server"); /** * Code for sending handshake */ + setState(State.CONNECTED); counter.set(1); + if (strategy != null) { - strategy.setAttemptsMade(0); + strategy.setNumberOfAttempts(0); } + try { + JSONObject handshakeObject = new JSONObject(); + handshakeObject.put("event", "#handshake"); + JSONObject object = new JSONObject(); + object.put("authToken", AuthToken); + handshakeObject.put("data", object); + handshakeObject.put("cid", counter.getAndIncrement()); + sendData(handshakeObject.toString()); - JSONObject handshakeObject = new JSONObject(); - handshakeObject.put("event", "#handshake"); - JSONObject object = new JSONObject(); - object.put("authToken", AuthToken); - handshakeObject.put("data", object); - handshakeObject.put("cid", counter.getAndIncrement()); - websocket.sendText(handshakeObject.toString()); + } catch (JSONException e) { + e.printStackTrace(); + } - listener.onConnected(Socket.this, headers); + listener.onConnected(Socket.this, response.headers()); + } - super.onConnected(websocket, headers); + @Override + public void onMessage(WebSocket webSocket, String text) { + try { + onTextMessage(text); + } catch (JSONException e) { + e.printStackTrace(); + } } @Override - public void onDisconnected(WebSocket websocket, WebSocketFrame serverCloseFrame, WebSocketFrame clientCloseFrame, boolean closedByServer) throws Exception { - listener.onDisconnected(Socket.this, serverCloseFrame, clientCloseFrame, closedByServer); - reconnect(); - super.onDisconnected(websocket, serverCloseFrame, clientCloseFrame, closedByServer); + public void onMessage(WebSocket webSocket, ByteString bytes) { + try { + onTextMessage(bytes.toString()); + } catch (JSONException e) { + e.printStackTrace(); + } } @Override - public void onConnectError(WebSocket websocket, WebSocketException exception) throws Exception { - listener.onConnectError(Socket.this, exception); - reconnect(); - super.onConnectError(websocket, exception); + public void onClosing(WebSocket webSocket, int code, String reason) { + logger.info("WebSocket closing: " + code + " - " + reason); + setState(State.DISCONNECTING); } + @Override + public void onClosed(WebSocket webSocket, int code, String reason) { + setState(State.DISCONNECTED); + logger.warning("Disconnected from server"); + pingHandler.removeLast(); + timeoutHandler.removeLast(); + processReconnection(); + listener.onDisconnected(Socket.this, code, reason); + + } @Override - public void onFrame(WebSocket websocket, WebSocketFrame frame) throws Exception { - - if (frame.getPayloadText().equalsIgnoreCase("#1")) { - /** - * PING-PONG logic goes here - */ - websocket.sendText("#2"); - } else { - - JSONObject object = new JSONObject(frame.getPayloadText()); - - /** - * Message retrieval mechanism goes here - */ - logger.info("Message :" + object.toString()); - - - try { - Object dataobject = object.opt("data"); - Integer rid = (Integer) object.opt("rid"); - Integer cid = (Integer) object.opt("cid"); - String event = (String) object.opt("event"); - - switch (Parser.parse(dataobject, event)) { - - case ISAUTHENTICATED: - listener.onAuthentication(Socket.this, ((JSONObject) dataobject).getBoolean("isAuthenticated")); - subscribeChannels(); - break; - case PUBLISH: - Socket.this.handlePublish(((JSONObject) dataobject).getString("channel"), ((JSONObject) dataobject).opt("data")); - break; - case REMOVETOKEN: - setAuthToken(null); - break; - case SETTOKEN: - String token = ((JSONObject) dataobject).getString("token"); - setAuthToken(token); - listener.onSetAuthToken(token, Socket.this); - break; - case EVENT: - if (hasEventAck(event)) { - handleEmitAck(event, dataobject, ack(Long.valueOf(cid))); - } else { - Socket.this.handleEmit(event, dataobject); + public void onFailure(WebSocket webSocket, Throwable throwable, Response response) { + logger.warning("Connect error: " + throwable); + setState(State.DISCONNECTED); + pingHandler.removeLast(); + timeoutHandler.removeLast(); + processReconnection(); + listener.onConnectError(Socket.this, throwable, response); + } + }; + } + + private void onTextMessage(String text) throws JSONException { + logger.info("Receiving: " + text); + + + // Valid message - reschedule next ping + reschedulePing(); + + // Proccess PING messages or send the message downstream + + if (text.equalsIgnoreCase(Constants.PING_MESSAGE)) { + sendData(Constants.PONG_MESSAGE); + } else { + JSONObject object = new JSONObject(text); + + /** + * Message retrieval mechanism goes here + */ + logger.info("Message :" + object.toString()); + + + try { + Object dataobject = object.opt("data"); + Integer rid = (Integer) object.opt("rid"); + Integer cid = (Integer) object.opt("cid"); + String event = (String) object.opt("event"); + + switch (Parser.parse(dataobject, event)) { + + case ISAUTHENTICATED: + listener.onAuthentication(this, ((JSONObject) dataobject).getBoolean("isAuthenticated")); + subscribeChannels(); + break; + case PUBLISH: + handlePublish(((JSONObject) dataobject).getString("channel"), ((JSONObject) dataobject).opt("data")); + break; + case REMOVETOKEN: + setAuthToken(null); + break; + case SETTOKEN: + String token = ((JSONObject) dataobject).getString("token"); + setAuthToken(token); + listener.onSetAuthToken(token, this); + break; + case EVENT: + if (hasEventAck(event)) { + handleEmitAck(event, dataobject, ack(Long.valueOf(cid))); + } else { + handleEmit(event, dataobject); + + } + break; + case ACKRECEIVE: + if (acks.containsKey((long) rid)) { + Object[] objects = acks.remove((long) rid); + if (objects != null) { + Ack fn = (Ack) objects[1]; + if (fn != null) { + fn.call((String) objects[0], object.opt("error"), object.opt("data")); + } else { + logger.warning("ack function is null with rid " + rid); } - break; - case ACKRECEIVE: - if (acks.containsKey((long) rid)) { - Object[] objects = acks.remove((long) rid); - if (objects != null) { - Ack fn = (Ack) objects[1]; - if (fn != null) { - fn.call((String) objects[0], object.opt("error"), object.opt("data")); - } else { - logger.warning("ack function is null with rid " + rid); - } - } - } - break; + } } - } catch (Exception e) { - logger.severe(e.toString()); - } - + break; } - super.onFrame(websocket, frame); + } catch (Exception e) { + logger.severe(e.toString()); } + } + } + public void setPingInterval(long pingInterval) { + pingEnable = true; + if (pingInterval != this.pingInterval) { + this.pingInterval = pingInterval; + } + } - @Override - public void onCloseFrame(WebSocket websocket, WebSocketFrame frame) throws Exception { - logger.warning("On close frame got called"); - super.onCloseFrame(websocket, frame); - } + public void disablePing() { + if (pingEnable) { + pingHandler.cancel(); + pingEnable = false; + } + } - @Override - public void onSendError(WebSocket websocket, WebSocketException cause, WebSocketFrame frame) throws Exception { - logger.severe("Error while sending data " + cause.toString()); - super.onSendError(websocket, cause, frame); - } + public void enablePing() { + if (!pingEnable) { + pingEnable = true; + sendData(Constants.PING_MESSAGE); + } + } - }; + public boolean isPingEnabled() { + return pingEnable; + } + private void setState(State state) { + logger.info(String.format("setState: old %s, new %s", currentState.name(), state.name())); + currentState = state; } public Socket emit(final String event, final Object object) { @@ -247,7 +299,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(eventObject.toString()); + sendData(eventObject.toString()); } }); return this; @@ -267,7 +319,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(eventObject.toString()); + sendData(eventObject.toString()); } }); return this; @@ -287,7 +339,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(subscribeObject.toString()); + sendData(subscribeObject.toString()); } }); return this; @@ -312,7 +364,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(subscribeObject.toString()); + sendData(subscribeObject.toString()); } }); return this; @@ -329,7 +381,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(subscribeObject.toString()); + sendData(subscribeObject.toString()); } }); return this; @@ -348,7 +400,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(subscribeObject.toString()); + sendData(subscribeObject.toString()); } }); return this; @@ -368,7 +420,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(publishObject.toString()); + sendData(publishObject.toString()); } }); @@ -390,7 +442,7 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(publishObject.toString()); + sendData(publishObject.toString()); } }); @@ -410,159 +462,159 @@ public void run() { } catch (JSONException e) { e.printStackTrace(); } - ws.sendText(object.toString()); + sendData(object.toString()); } }); } }; } - private void subscribeChannels() { - for(Map.Entry entry : channels.entrySet()) { + for (Map.Entry entry : channels.entrySet()) { entry.getValue().subscribe(); } } public void setExtraHeaders(Map extraHeaders, boolean overrideDefaultHeaders) { + Headers.Builder builder = new Headers.Builder(); if (overrideDefaultHeaders) { - headers.clear(); + builder = getDefaultHeadersBuilder(); } - - headers.putAll(extraHeaders); + for (Map.Entry entry : extraHeaders.entrySet()) { + builder.add(entry.getKey(), entry.getValue()); + } + headers = builder.build(); } - public Map getHeaders() { + public Headers getHeaders() { return headers; } - public void connect() { + protected void sendDataInBackground(String message) { + sendData(message); + } - try { - ws = factory.createSocket(URL); - } catch (IOException e) { - logger.severe(e.toString()); - } - ws.addExtension("permessage-deflate; client_max_window_bits"); - for (Map.Entry entry : headers.entrySet()) { - ws.addHeader(entry.getKey(), entry.getValue()); + public void sendData(String message) { + if (getState() == State.CONNECTED) { + logger.info("Sending: " + message); + ws.send(message); } + } - ws.addListener(adapter); - - try { - ws.connect(); - } catch (OpeningHandshakeException e) { - // A violation against the WebSocket protocol was detected - // during the opening handshake. - - logger.severe(e.toString()); - // Status line. - StatusLine sl = e.getStatusLine(); - logger.info("=== Status Line ==="); - logger.info("HTTP Version = \n" + sl.getHttpVersion()); - logger.info("Status Code = \n" + sl.getStatusCode()); - logger.info("Reason Phrase = \n" + sl.getReasonPhrase()); - - // HTTP headers. - Map> headers = e.getHeaders(); - logger.info("=== HTTP Headers ==="); - for (Map.Entry> entry : headers.entrySet()) { - // Header name. - String name = entry.getKey(); - - // Values of the header. - List values = entry.getValue(); - - if (values == null || values.size() == 0) { - // Print the name only. - logger.info(name); - continue; - } - for (String value : values) { - // Print the name and the value. - logger.info(name + value + "\n"); - } - } - } catch (WebSocketException e) { - listener.onConnectError(Socket.this, e); - reconnect(); - } + protected Request buildRequest(Headers headers) { + setState(State.CREATED); + // Create a WebSocket with a socket connection timeout value. + return new Request.Builder() + .url(url) + .headers(headers) + .build(); + } + private void connect() { + Request request = buildRequest(headers); + setState(State.CONNECTING); + ws = client.newWebSocket(request, getWebscoketListener()); } public void connectAsync() { - try { - ws = factory.createSocket(URL); - } catch (IOException e) { - logger.severe(e.toString()); - } - ws.addExtension("permessage-deflate; client_max_window_bits"); - for (Map.Entry entry : headers.entrySet()) { - ws.addHeader(entry.getKey(), entry.getValue()); - } + connect(); + } - ws.addListener(adapter); - ws.connectAsynchronously(); + public void reconnect() { + logger.info("reconnecting"); + connect(); } - private void reconnect() { - if (strategy == null) { - logger.warning("Unable to reconnect: reconnection is null"); + + public void disconnect() { + disconnect("close"); + } + + public void disconnect(String reason) { + logger.info("Calling disconnect"); + if (currentState == State.DISCONNECTED) { return; + } else if (currentState == State.CONNECTED) { + ws.close(1001, reason); + setState(State.DISCONNECTING); + } else { + setState(State.DISCONNECTED); } - if (strategy.areAttemptsComplete()) { - strategy.setAttemptsMade(0); - logger.warning("Unable to reconnect: max reconnection attempts reached"); - return; + pingHandler.removeLast(); + timeoutHandler.removeLast(); + selfDisconnect = true; + } + + + /* visible for testing */ + void processReconnection() { + if (strategy != null && !selfDisconnect) { + if (strategy.getNumberOfAttempts() < strategy.getMaxAttempts()) { + timer = new Timer(); + timer.schedule(new TimerTask() { + @Override + public void run() { + reconnect(); + strategy.processAttempts(); + timer.cancel(); + timer.purge(); + } + }, strategy.getReconnectInterval()); + + } else { + pingHandler.cancel(); + logger.info("Number of attempts are complete"); + } + } else { + pingHandler.cancel(); + selfDisconnect = false; } + } - final Timer timer = new Timer(); - timer.schedule(new TimerTask() { + // TODO: 15/8/17 solve problem of PONG RECEIVE FAILED by giving a fair chance + protected void reschedulePing() { + if (!pingEnable) + return; + + logger.info("Scheduling ping in: " + pingInterval + " ms"); + pingHandler.removeLast(); + timeoutHandler.removeLast(); + pingHandler.postDelayed(new TimerTask() { @Override public void run() { - if (strategy == null) { - logger.warning("Unable to reconnect: reconnection is null"); - return; + logger.info("SENDING PING"); + sendData(Constants.PING_MESSAGE); + } + }, pingInterval); + timeoutHandler.postDelayed(new TimerTask() { + @Override + public void run() { + if (getState() != State.DISCONNECTING && getState() != State.DISCONNECTED) { + logger.warning("PONG RECEIVE FAILED"); + ws.cancel(); + //onFailure(ws, new IOException("PING Timeout"), null); } - strategy.processValues(); - Socket.this.connect(); - timer.cancel(); - timer.purge(); + timeoutHandler.removeLast(); } - }, strategy.getReconnectInterval()); - } - - public void disconnect() { - if (ws != null) { - ws.disconnect(); - } - strategy = null; + }, 2 * pingInterval); } - /** - * States can be - * CLOSED - * CLOSING - * CONNECTING - * CREATED - * OPEN - */ - public WebSocketState getCurrentState() { - return ws != null ? ws.getState() : null; + public State getState() { + return currentState; } public Boolean isconnected() { - return ws != null && ws.getState() == WebSocketState.OPEN; + return getState() == State.CONNECTED; } public void disableLogging() { logger.setLevel(Level.OFF); } + /** * Channels need to be subscribed everytime whenever client is reconnected to server (handled inside) * Add only one listener to one channel for whole lifetime of process @@ -613,7 +665,9 @@ public void unsubscribe(Ack ack) { @Override protected void finalize() throws Throwable { - ws.disconnect("Client socket garbage collected, closing connection"); + disconnect("Client socket garbage collected, closing connection"); super.finalize(); } + } + diff --git a/src/main/java/io/github/sac/State.java b/src/main/java/io/github/sac/State.java new file mode 100644 index 0000000..bb6c6f0 --- /dev/null +++ b/src/main/java/io/github/sac/State.java @@ -0,0 +1,9 @@ +package io.github.sac; + +public enum State { + CREATED, + CONNECTING, + CONNECTED, + DISCONNECTING, + DISCONNECTED, +} diff --git a/src/main/java/io/github/sac/TaskHandler.java b/src/main/java/io/github/sac/TaskHandler.java new file mode 100644 index 0000000..3893f88 --- /dev/null +++ b/src/main/java/io/github/sac/TaskHandler.java @@ -0,0 +1,57 @@ +package io.github.sac; + +import java.util.Timer; +import java.util.TimerTask; + +/** + * Created by sachin on 4/8/17. + */ +public class TaskHandler { + private Timer timer; + private TimerTask task; + private Boolean isCancelled; + + public TaskHandler() { + timer = new Timer(); + isCancelled = false; + } + + public void postDelayed(TimerTask timerTask, long delay) { + this.task = timerTask; + if (isCancelled) { + recreate(); + } + timer.schedule(timerTask, delay); + } + + public void scheduleAtFixedRate(TimerTask timerTask, long delay, long period) { + if (isCancelled) { + recreate(); + } + timer.scheduleAtFixedRate(timerTask, delay, period); + } + + public void removeLast() { + if (task != null) { + task.cancel(); + } + } + + public void remove(TimerTask task) { + task.cancel(); + } + + public void cancel() { + if (!isCancelled) { + removeLast(); + timer.cancel(); + timer.purge(); + isCancelled = true; + } + } + + private void recreate() { + timer = new Timer(); + isCancelled = false; + } +}