diff --git a/src/main/java/io/github/sac/Emitter.java b/src/main/java/io/github/sac/Emitter.java index cdf49b1..bb585f8 100644 --- a/src/main/java/io/github/sac/Emitter.java +++ b/src/main/java/io/github/sac/Emitter.java @@ -4,15 +4,43 @@ * Created by sachin on 13/11/16. */ +import java.util.Iterator; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; public class Emitter { + public static String RAWEVENT = "raw"; + public static String MESSAGEEVENT = "message"; - private ConcurrentHashMap singlecallbacks = new ConcurrentHashMap<>(); - private ConcurrentHashMap singleackcallbacks = new ConcurrentHashMap<>(); - private ConcurrentHashMap publishcallbacks = new ConcurrentHashMap<>(); + private boolean multipleListenersEnabled; + private boolean multipleChannelWatchersEnabled; + + public Emitter(boolean multipleListenersEnabled, boolean multipleChannelWatchersEnabled) { + this.multipleListenersEnabled = multipleListenersEnabled; + this.multipleChannelWatchersEnabled = multipleChannelWatchersEnabled; + } + + public void setMultipleListenersEnabled(boolean multipleListenersEnabled) { + this.multipleListenersEnabled = multipleListenersEnabled; + } + + public boolean isMultipleListenersEnabled() { + return multipleListenersEnabled; + } + + public boolean isMultipleChannelWatchersEnabled() { + return multipleChannelWatchersEnabled; + } + + public void setMultipleChannelWatchersEnabled(boolean multipleChannelWatchersEnabled) { + this.multipleChannelWatchersEnabled = multipleChannelWatchersEnabled; + } + + private ConcurrentHashMap> listeners = new ConcurrentHashMap<>(); + private ConcurrentHashMap> ackListeners = new ConcurrentHashMap<>(); + private ConcurrentHashMap> channelObservers = new ConcurrentHashMap<>(); /** * Listens on the event. @@ -20,98 +48,131 @@ public class Emitter { * @param event event name. * @return a reference to this object. */ - public Emitter on(String event, Listener fn) { - if (singlecallbacks.containsKey(event)) { - singlecallbacks.remove(event); - } - singlecallbacks.put(event, fn); - return this; + public Emitter onRawEvent(Listener fn) { + return on(RAWEVENT, fn); } - public Emitter onSubscribe(String event, Listener fn) { + public Emitter onAnyMessage(Listener fn) { + return on(MESSAGEEVENT, fn); + } - if (publishcallbacks.containsKey(event)) { - publishcallbacks.remove(event); - } - publishcallbacks.put(event, fn); + public Emitter onAnyMessage(AckListener fn) { + return on(MESSAGEEVENT, fn); + } + + public Emitter on(String event, Listener fn) { + return on(event, fn, multipleListenersEnabled); + } + + public Emitter on(String event, Listener fn, boolean multiListenersEnabled) { + registerEvent(listeners, event, fn, multiListenersEnabled); return this; } public Emitter on(String event, AckListener fn) { - if (singleackcallbacks.containsKey(event)) { - singleackcallbacks.remove(event); - } - singleackcallbacks.put(event, fn); - return this; + return on(event, fn, multipleListenersEnabled); } + public Emitter on(String event, AckListener fn, boolean multiListenersEnabled) { + registerEvent(ackListeners, event, fn, multiListenersEnabled); + return this; + } - public Emitter handleEmit(String event, Object object) { + public Emitter onSubscribe(String event, Listener fn) { + return onSubscribe(event, fn, multipleChannelWatchersEnabled); + } - Listener listener = singlecallbacks.get(event); - if (listener != null) { - listener.call(event, object); - } + public Emitter onSubscribe(String event, Listener fn, boolean multipleChannelWatchersEnabled) { + registerEvent(channelObservers, event, fn, multipleChannelWatchersEnabled); return this; } - public Emitter handlePublish(String event, Object object) { - - Listener listener = publishcallbacks.get(event); - if (listener != null) { - listener.call(event, object); + private static void registerEvent(ConcurrentHashMap> listeners, String event, T fn, boolean multiEnabled) { + if (listeners.containsKey(event)) { + if (!multiEnabled) { + listeners.get(event).clear(); + } + listeners.get(event).add(fn); + return; } - return this; + ConcurrentLinkedQueue linkedListeners = new ConcurrentLinkedQueue<>(); + linkedListeners.add(fn); + listeners.put(event, linkedListeners); } - public boolean hasEventAck(String event) { - return this.singleackcallbacks.get(event) != null; + + public Emitter handleEmit(String event, Object object) { + handleEvent(listeners, event, object, null); + return this; } + public Emitter handleEmitAck(String event, Object object, Ack ack) { + handleEvent(ackListeners, event, object, ack); + return this; + } - AckListener listener = singleackcallbacks.get(event); - if (listener != null) { - listener.call(event, object, ack); - } + public Emitter handlePublish(String event, Object object) { + handleEvent(channelObservers, event, object, null); return this; } - public interface Listener { - void call(String name, Object data); + public static void handleEvent(ConcurrentHashMap> listeners, String event, Object object, Ack ack) { + InvokeListeners(event, object, ack, listeners.get(event).iterator()); + InvokeListeners(event, object, ack, listeners.get(MESSAGEEVENT).iterator()); } - public interface AckListener { - void call(String name, Object data, Ack ack); + public static void InvokeListeners(String event, Object object, Ack ack, Iterator listenerIterator) { + while (listenerIterator.hasNext()) { + T listener = listenerIterator.next(); + if (listener instanceof Listener) { + ((Listener) listener).call(event, object); + } else { + ((AckListener) listener).call(event, object, ack); + } + } } - /** - * New methods ADDED - */ + public void off(String event) { + listeners.remove(event); + ackListeners.remove(event); + } - public void removeEmitCallback(String event) { - singlecallbacks.remove(event); - singleackcallbacks.remove(event); + public void off(String event, Listener listener) { + if (listeners.containsKey(event)) { + listeners.get(event).remove(listener); + } } - public void removeSubscribeCallback(String event) { - publishcallbacks.remove(event); + public void off(String event, AckListener ackListener) { + if (ackListeners.containsKey(event)) { + ackListeners.get(event).remove(ackListener); + } } - public void removeAllCallbacks() { - for (Map.Entry e : singlecallbacks.entrySet()) { - singlecallbacks.remove(e.getKey().toString()); + public void removeAllListeners() { + for (Map.Entry e : listeners.entrySet()) { + listeners.remove(e.getKey().toString()); } - for (Map.Entry e : singleackcallbacks.entrySet()) { - singleackcallbacks.remove(e.getKey().toString()); + for (Map.Entry e : ackListeners.entrySet()) { + ackListeners.remove(e.getKey().toString()); } - for (Map.Entry e : publishcallbacks.entrySet()) { - publishcallbacks.remove(e.getKey().toString()); + for (Map.Entry e : channelObservers.entrySet()) { + channelObservers.remove(e.getKey().toString()); } } + + public interface Listener { + void call(String name, Object data); + } + + public interface AckListener { + void call(String name, Object data, Ack ack); + } + } diff --git a/src/main/java/io/github/sac/Parser.java b/src/main/java/io/github/sac/Parser.java index b0f8005..5877c78 100644 --- a/src/main/java/io/github/sac/Parser.java +++ b/src/main/java/io/github/sac/Parser.java @@ -9,6 +9,7 @@ */ public class Parser { + // todo : Probably need to add SUBSCRIBE AND UNSUBSCRIBE EVENTS FROM SERVER IN PARSERESULT public enum ParseResult { ISAUTHENTICATED, PUBLISH, diff --git a/src/main/java/io/github/sac/Socket.java b/src/main/java/io/github/sac/Socket.java index f8c9c0b..0ffe7ef 100644 --- a/src/main/java/io/github/sac/Socket.java +++ b/src/main/java/io/github/sac/Socket.java @@ -4,10 +4,15 @@ import com.neovisionaries.ws.client.StatusLine; import com.neovisionaries.ws.client.WebSocket; import com.neovisionaries.ws.client.WebSocketAdapter; +import com.neovisionaries.ws.client.WebSocketCloseCode; 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 io.github.sac.events.AuthenticationEvent; +import io.github.sac.events.ChannelKickoutEvent; +import io.github.sac.events.ErrorEvent; +import io.github.sac.events.SubscribeStateEvent; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; @@ -40,9 +45,37 @@ public class Socket extends Emitter { private List channels; private WebSocketAdapter adapter; private Map headers; + private AuthState authState; + + // Extra definition of events + public AuthenticationEvent authenticationEventHandler; + public ChannelKickoutEvent channelKickoutEventHandler; + public ErrorEvent errorEventHandler; + public SubscribeStateEvent subscribeStateEventHandler; + public Socket(String URL) { + this(URL, null); + } + + public Socket(String URL, BasicListener listener) { + this(URL, listener, null); + } + + public Socket(String URL, BasicListener listener, String authToken) { + this(URL, listener, authToken, null); + } + + public Socket(String URL, BasicListener listener, String authToken, ReconnectStrategy reconnectStrategy) { + this(URL, listener, authToken, reconnectStrategy, false, false); + } + + public Socket(String URL, BasicListener listener, String AuthToken, ReconnectStrategy reconnectStrategy, boolean multipleListenersEnabled, boolean multipleChannelWatchersEnabled) { + super(multipleListenersEnabled, multipleChannelWatchersEnabled); this.URL = URL; + this.listener = listener; + this.AuthToken = AuthToken; + strategy = reconnectStrategy; factory = new WebSocketFactory().setConnectionTimeout(5000); counter = new AtomicInteger(1); acks = new HashMap<>(); @@ -60,7 +93,11 @@ private void putDefaultHeaders() { } public Channel createChannel(String name) { - Channel channel = new Channel(name); + return createChannel(name, true); + } + + public Channel createChannel(String name, boolean autoSubscribe) { + Channel channel = new Channel(name, autoSubscribe); channels.add(channel); return channel; } @@ -69,6 +106,18 @@ public List getChannels() { return channels; } + public void subscribeAllChannels() { + for (Channel channel : channels) { + channel.subscribe(); + } + } + + public void unsubscribeAllChannels() { + for (Channel channel : channels) { + channel.unsubscribe(); + } + } + public Channel getChannelByName(String name) { for (Channel channel : channels) { if (channel.getChannelName().equals(name)) @@ -89,9 +138,10 @@ public void setListener(BasicListener listener) { this.listener = listener; } - public Logger getLogger(){ + public Logger getLogger() { return logger; } + /** * used to set up TLS/SSL connection to server for more details visit neovisionaries websocket client */ @@ -104,6 +154,14 @@ public void setAuthToken(String token) { AuthToken = token; } + public String getAuthToken() { + return AuthToken; + } + + public AuthState getAuthState() { + return authState; + } + public WebSocketAdapter getAdapter() { return new WebSocketAdapter() { @@ -164,7 +222,6 @@ public void onFrame(WebSocket websocket, WebSocketFrame frame) throws Exception */ logger.info("Message :" + object.toString()); - try { Object dataobject = object.opt("data"); Integer rid = (Integer) object.opt("rid"); @@ -174,8 +231,10 @@ public void onFrame(WebSocket websocket, WebSocketFrame frame) throws Exception switch (Parser.parse(dataobject, event)) { case ISAUTHENTICATED: - listener.onAuthentication(Socket.this, ((JSONObject) dataobject).getBoolean("isAuthenticated")); - subscribeChannels(); + boolean isAuthenticated = ((JSONObject) dataobject).getBoolean("isAuthenticated"); + authState = isAuthenticated ? AuthState.AUTHENTICATED : AuthState.UNAUTHENTICATED; + listener.onAuthentication(Socket.this, isAuthenticated); + subscribeAllChannels(); break; case PUBLISH: Socket.this.handlePublish(((JSONObject) dataobject).getString("channel"), ((JSONObject) dataobject).opt("data")); @@ -189,12 +248,8 @@ public void onFrame(WebSocket websocket, WebSocketFrame frame) throws Exception listener.onSetAuthToken(token, Socket.this); break; case EVENT: - if (hasEventAck(event)) { - handleEmitAck(event, dataobject, ack(Long.valueOf(cid))); - } else { - Socket.this.handleEmit(event, dataobject); - - } + handleEmitAck(event, dataobject, ack(Long.valueOf(cid))); + handleEmit(event, dataobject); break; case ACKRECEIVE: if (acks.containsKey((long) rid)) { @@ -211,6 +266,9 @@ public void onFrame(WebSocket websocket, WebSocketFrame frame) throws Exception break; } } catch (Exception e) { + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } logger.severe(e.toString()); } @@ -235,6 +293,10 @@ public void onSendError(WebSocket websocket, WebSocketException cause, WebSocket } + public Socket send(final Object object) { + return emit(RAWEVENT, object); + } + public Socket emit(final String event, final Object object) { EventThread.exec(new Runnable() { public void run() { @@ -244,6 +306,9 @@ public void run() { eventObject.put("data", object); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(eventObject.toString()); } @@ -264,6 +329,9 @@ public void run() { eventObject.put("cid", counter.getAndIncrement()); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(eventObject.toString()); } @@ -284,6 +352,9 @@ public void run() { subscribeObject.put("cid", counter.getAndIncrement()); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(subscribeObject.toString()); } @@ -309,6 +380,9 @@ public void run() { subscribeObject.put("cid", counter.getAndIncrement()); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(subscribeObject.toString()); } @@ -326,6 +400,9 @@ public void run() { subscribeObject.put("cid", counter.getAndIncrement()); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(subscribeObject.toString()); } @@ -345,6 +422,9 @@ public void run() { subscribeObject.put("cid", counter.getAndIncrement()); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(subscribeObject.toString()); } @@ -365,6 +445,9 @@ public void run() { publishObject.put("cid", counter.getAndIncrement()); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(publishObject.toString()); } @@ -387,6 +470,9 @@ public void run() { publishObject.put("cid", counter.getAndIncrement()); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(publishObject.toString()); } @@ -407,6 +493,9 @@ public void run() { object.put("rid", cid); } catch (JSONException e) { e.printStackTrace(); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.sendText(object.toString()); } @@ -416,12 +505,6 @@ public void run() { } - private void subscribeChannels() { - for (Channel channel : channels) { - channel.subscribe(); - } - } - public void setExtraHeaders(Map extraHeaders, boolean overrideDefaultHeaders) { if (overrideDefaultHeaders) { headers.clear(); @@ -435,17 +518,7 @@ public Map getHeaders() { } public void connect() { - - 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()); - } - + CreateSocket(); ws.addListener(adapter); try { @@ -453,6 +526,9 @@ public void connect() { } catch (OpeningHandshakeException e) { // A violation against the WebSocket protocol was detected // during the opening handshake. + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } logger.severe(e.toString()); // Status line. @@ -485,22 +561,31 @@ public void connect() { } } catch (WebSocketException e) { listener.onConnectError(Socket.this, e); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } reconnect(); } } - public void connectAsync() { + private void CreateSocket() { try { ws = factory.createSocket(URL); } catch (IOException e) { logger.severe(e.toString()); + if (errorEventHandler != null) { + errorEventHandler.onError(Socket.this, e); + } } ws.addExtension("permessage-deflate; client_max_window_bits"); for (Map.Entry entry : headers.entrySet()) { ws.addHeader(entry.getKey(), entry.getValue()); } + } + public void connectAsync() { + CreateSocket(); ws.addListener(adapter); ws.connectAsynchronously(); } @@ -534,10 +619,23 @@ public void run() { } public void disconnect() { + disconnect(WebSocketCloseCode.NORMAL, null); + } + + public void disconnect(String closeReason) { + disconnect(WebSocketCloseCode.NORMAL, closeReason); + } + + public void disconnect(int closeCode, String closeReason) { + disconnect(closeCode, closeReason, -1); + } + + public void disconnect(int closeCode, String closeReason, long closeDelay) { + unsubscribeAllChannels(); + strategy.setAttemptsMade(strategy.maxAttempts); if (ws != null) { - ws.disconnect(); + ws.disconnect(closeCode, closeReason, closeDelay); } - strategy = null; } /** @@ -549,10 +647,34 @@ public void disconnect() { * OPEN */ - public WebSocketState getCurrentState() { + public WebSocketState getSocketStatus() { return ws.getState(); } + public SocketState getConnectionState() { + switch (getSocketStatus()) { + case CREATED: + return SocketState.CREATED; + case CONNECTING: + return SocketState.CONNECTING; + case OPEN: + return SocketState.OPEN; + case CLOSING: + return SocketState.CLOSING; + case CLOSED: + return SocketState.CLOSED; + } + + switch (getAuthState()) { + case AUTHENTICATED: + return SocketState.AUTHENTICATED; + case UNAUTHENTICATED: + return SocketState.UNAUTHENTICATED; + } + return SocketState.NOTFOUND; + } + + public Boolean isconnected() { return ws != null && ws.getState() == WebSocketState.OPEN; } @@ -569,13 +691,17 @@ public void disableLogging() { public class Channel { String channelName; + boolean autoSubscribe; + ChannelState channelState; public String getChannelName() { return channelName; } - public Channel(String channelName) { + public Channel(String channelName, boolean autoSubscribe) { this.channelName = channelName; + this.autoSubscribe = autoSubscribe; + this.channelState = ChannelState.UNSUBSCRIBED; } public void subscribe() { @@ -600,15 +726,51 @@ public void publish(Object data, Ack ack) { public void unsubscribe() { Socket.this.unsubscribe(channelName); - channels.remove(this); } public void unsubscribe(Ack ack) { Socket.this.unsubscribe(channelName, ack); - channels.remove(this); } } + public enum AuthState { + AUTHENTICATED, + UNAUTHENTICATED + } + + public enum SocketState { + CREATED, + CONNECTING, + OPEN, + CLOSING, + CLOSED, + AUTHENTICATED, + UNAUTHENTICATED, + NOTFOUND + } + + public enum ChannelState { + SUBSCRIBED, + PENDING, + UNSUBSCRIBED + } + + public void setAuthenticationEventHandler(AuthenticationEvent authenticationEventHandler) { + this.authenticationEventHandler = authenticationEventHandler; + } + + public void setChannelKickoutEventHandler(ChannelKickoutEvent channelKickoutEventHandler) { + this.channelKickoutEventHandler = channelKickoutEventHandler; + } + + public void setErrorEventHandler(ErrorEvent errorEventHandler) { + this.errorEventHandler = errorEventHandler; + } + + public void setSubscribeStateEventHandler(SubscribeStateEvent subscribeStateEventHandler) { + this.subscribeStateEventHandler = subscribeStateEventHandler; + } + @Override protected void finalize() throws Throwable { ws.disconnect("Client socket garbage collected, closing connection"); diff --git a/src/main/java/io/github/sac/events/AuthenticationEvent.java b/src/main/java/io/github/sac/events/AuthenticationEvent.java new file mode 100644 index 0000000..459b811 --- /dev/null +++ b/src/main/java/io/github/sac/events/AuthenticationEvent.java @@ -0,0 +1,11 @@ +package io.github.sac.events; + +import io.github.sac.Socket; + +/** + * Created by sachin on 3/7/18. + */ +public interface AuthenticationEvent { + void onAuthenticated(Socket socket, String token); + void onDeauthentication(Socket socket); +} diff --git a/src/main/java/io/github/sac/events/ChannelKickoutEvent.java b/src/main/java/io/github/sac/events/ChannelKickoutEvent.java new file mode 100644 index 0000000..aa0749d --- /dev/null +++ b/src/main/java/io/github/sac/events/ChannelKickoutEvent.java @@ -0,0 +1,10 @@ +package io.github.sac.events; + +import io.github.sac.Socket; + +/** + * Created by sachin on 3/7/18. + */ +public interface ChannelKickoutEvent { + void onChannelKickout(Socket socket, String message, String channelName); +} diff --git a/src/main/java/io/github/sac/events/ErrorEvent.java b/src/main/java/io/github/sac/events/ErrorEvent.java new file mode 100644 index 0000000..c8f7d9f --- /dev/null +++ b/src/main/java/io/github/sac/events/ErrorEvent.java @@ -0,0 +1,11 @@ +package io.github.sac.events; + +import io.github.sac.Socket; + +/** + * Created by sachin on 3/7/18. + */ + +public interface ErrorEvent { + void onError(Socket socket, Exception error); +} diff --git a/src/main/java/io/github/sac/events/SubscribeStateEvent.java b/src/main/java/io/github/sac/events/SubscribeStateEvent.java new file mode 100644 index 0000000..ee4437b --- /dev/null +++ b/src/main/java/io/github/sac/events/SubscribeStateEvent.java @@ -0,0 +1,10 @@ +package io.github.sac.events; + +import io.github.sac.Socket; + +/** + * Created by sachin on 3/7/18. + */ +public interface SubscribeStateEvent { + void onSubscribeStateChange(Socket.Channel channel, Socket.ChannelState oldState, Socket.ChannelState newState); +}