managers = new ConcurrentHashMap<>();
/**
* Protocol version.
@@ -58,20 +57,20 @@ public static Socket socket(URI uri, Options opts) {
opts = new Options();
}
- URL parsed = Url.parse(uri);
- URI source;
- try {
- source = parsed.toURI();
- } catch (URISyntaxException e) {
- throw new RuntimeException(e);
- }
- String id = Url.extractId(parsed);
- String path = parsed.getPath();
+ Url.ParsedURI parsed = Url.parse(uri);
+ URI source = parsed.uri;
+ String id = parsed.id;
+
boolean sameNamespace = managers.containsKey(id)
- && managers.get(id).nsps.containsKey(path);
+ && managers.get(id).nsps.containsKey(source.getPath());
boolean newConnection = opts.forceNew || !opts.multiplex || sameNamespace;
Manager io;
+ String query = source.getQuery();
+ if (query != null && (opts.query == null || opts.query.isEmpty())) {
+ opts.query = query;
+ }
+
if (newConnection) {
if (logger.isLoggable(Level.FINE)) {
logger.fine(String.format("ignoring socket cache for %s", source));
@@ -87,12 +86,7 @@ public static Socket socket(URI uri, Options opts) {
io = managers.get(id);
}
- String query = parsed.getQuery();
- if (query != null && (opts.query == null || opts.query.isEmpty())) {
- opts.query = query;
- }
-
- return io.socket(parsed.getPath(), opts);
+ return io.socket(source.getPath(), opts);
}
@@ -104,5 +98,21 @@ public static class Options extends Manager.Options {
* Whether to enable multiplexing. Default is true.
*/
public boolean multiplex = true;
+
+ /**
+ *
+ * Retrieve new builder class that helps creating socket option as builder pattern.
+ * This method returns exactly same result as :
+ *
+ *
+ * SocketOptionBuilder builder = SocketOptionBuilder.builder();
+ *
+ *
+ * @return builder class that helps creating socket option as builder pattern.
+ * @see SocketOptionBuilder#builder()
+ */
+ public static SocketOptionBuilder builder() {
+ return SocketOptionBuilder.builder();
+ }
}
}
diff --git a/src/main/java/io/socket/client/Manager.java b/src/main/java/io/socket/client/Manager.java
index 1058b067..04198998 100644
--- a/src/main/java/io/socket/client/Manager.java
+++ b/src/main/java/io/socket/client/Manager.java
@@ -2,6 +2,7 @@
import io.socket.backo.Backoff;
import io.socket.emitter.Emitter;
+import io.socket.parser.DecodingException;
import io.socket.parser.IOParser;
import io.socket.parser.Packet;
import io.socket.parser.Parser;
@@ -10,16 +11,7 @@
import okhttp3.WebSocket;
import java.net.URI;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.Set;
-import java.util.Timer;
-import java.util.TimerTask;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -48,16 +40,6 @@ public class Manager extends Emitter {
public static final String EVENT_PACKET = "packet";
public static final String EVENT_ERROR = "error";
- /**
- * Called on a connection error.
- */
- public static final String EVENT_CONNECT_ERROR = "connect_error";
-
- /**
- * Called on a connection timeout.
- */
- public static final String EVENT_CONNECT_TIMEOUT = "connect_timeout";
-
/**
* Called on a successful reconnection.
*/
@@ -72,12 +54,6 @@ public class Manager extends Emitter {
public static final String EVENT_RECONNECT_ATTEMPT = "reconnect_attempt";
- public static final String EVENT_RECONNECTING = "reconnecting";
-
- public static final String EVENT_PING = "ping";
-
- public static final String EVENT_PONG = "pong";
-
/**
* Called when a new transport is created. (experimental)
*/
@@ -96,22 +72,20 @@ public class Manager extends Emitter {
private long _reconnectionDelay;
private long _reconnectionDelayMax;
private double _randomizationFactor;
- private Backoff backoff;
+ private final Backoff backoff;
private long _timeout;
- private Set connecting = new HashSet();
- private Date lastPing;
- private URI uri;
- private List packetBuffer;
- private Queue subs;
- private Options opts;
+ private final URI uri;
+ private final List packetBuffer = new ArrayList<>();
+ private final Queue subs = new LinkedList<>();;
+ private final Options opts;
/*package*/ io.socket.engineio.client.Socket engine;
- private Parser.Encoder encoder;
- private Parser.Decoder decoder;
+ private final Parser.Encoder encoder;
+ private final Parser.Decoder decoder;
/**
* This HashMap can be accessed from outside of EventThread.
*/
- /*package*/ ConcurrentHashMap nsps;
+ /*package*/ final Map nsps = new ConcurrentHashMap<>();
public Manager() {
@@ -140,8 +114,6 @@ public Manager(URI uri, Options opts) {
opts.callFactory = defaultCallFactory;
}
this.opts = opts;
- this.nsps = new ConcurrentHashMap();
- this.subs = new LinkedList();
this.reconnection(opts.reconnection);
this.reconnectionAttempts(opts.reconnectionAttempts != 0 ? opts.reconnectionAttempts : Integer.MAX_VALUE);
this.reconnectionDelay(opts.reconnectionDelay != 0 ? opts.reconnectionDelay : 1000);
@@ -155,33 +127,10 @@ public Manager(URI uri, Options opts) {
this.readyState = ReadyState.CLOSED;
this.uri = uri;
this.encoding = false;
- this.packetBuffer = new ArrayList();
this.encoder = opts.encoder != null ? opts.encoder : new IOParser.Encoder();
this.decoder = opts.decoder != null ? opts.decoder : new IOParser.Decoder();
}
- private void emitAll(String event, Object... args) {
- this.emit(event, args);
- for (Socket socket : this.nsps.values()) {
- socket.emit(event, args);
- }
- }
-
- /**
- * Update `socket.id` of all sockets
- */
- private void updateSocketIds() {
- for (Map.Entry entry : this.nsps.entrySet()) {
- String nsp = entry.getKey();
- Socket socket = entry.getValue();
- socket.id = this.generateId(nsp);
- }
- }
-
- private String generateId(String nsp) {
- return ("/".equals(nsp) ? "" : (nsp + "#")) + this.engine.id();
- }
-
public boolean reconnection() {
return this._reconnection;
}
@@ -307,7 +256,7 @@ public void call(Object... objects) {
logger.fine("connect_error");
self.cleanup();
self.readyState = ReadyState.CLOSED;
- self.emitAll(EVENT_CONNECT_ERROR, data);
+ self.emit(EVENT_ERROR, data);
if (fn != null) {
Exception err = new SocketIOException("Connection error",
data instanceof Exception ? (Exception) data : null);
@@ -319,24 +268,28 @@ public void call(Object... objects) {
}
});
- if (Manager.this._timeout >= 0) {
- final long timeout = Manager.this._timeout;
+ final long timeout = Manager.this._timeout;
+ final Runnable onTimeout = new Runnable() {
+ @Override
+ public void run() {
+ logger.fine(String.format("connect attempt timed out after %d", timeout));
+ openSub.destroy();
+ socket.close();
+ socket.emit(Engine.EVENT_ERROR, new SocketIOException("timeout"));
+ }
+ };
+
+ if (timeout == 0) {
+ EventThread.exec(onTimeout);
+ return;
+ } else if (Manager.this._timeout > 0) {
logger.fine(String.format("connection attempt will timeout after %d", timeout));
final Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
- EventThread.exec(new Runnable() {
- @Override
- public void run() {
- logger.fine(String.format("connect attempt timed out after %d", timeout));
- openSub.destroy();
- socket.close();
- socket.emit(Engine.EVENT_ERROR, new SocketIOException("timeout"));
- self.emitAll(EVENT_CONNECT_TIMEOUT, timeout);
- }
- });
+ EventThread.exec(onTimeout);
}
}, timeout);
@@ -370,25 +323,17 @@ private void onopen() {
@Override
public void call(Object... objects) {
Object data = objects[0];
- if (data instanceof String) {
- Manager.this.ondata((String)data);
- } else if (data instanceof byte[]) {
- Manager.this.ondata((byte[])data);
+ try {
+ if (data instanceof String) {
+ Manager.this.decoder.add((String) data);
+ } else if (data instanceof byte[]) {
+ Manager.this.decoder.add((byte[]) data);
+ }
+ } catch (DecodingException e) {
+ logger.fine("error while decoding the packet: " + e.getMessage());
}
}
}));
- this.subs.add(On.on(socket, Engine.EVENT_PING, new Listener() {
- @Override
- public void call(Object... objects) {
- Manager.this.onping();
- }
- }));
- this.subs.add(On.on(socket, Engine.EVENT_PONG, new Listener() {
- @Override
- public void call(Object... objects) {
- Manager.this.onpong();
- }
- }));
this.subs.add(On.on(socket, Engine.EVENT_ERROR, new Listener() {
@Override
public void call(Object... objects) {
@@ -409,31 +354,13 @@ public void call (Packet packet) {
});
}
- private void onping() {
- this.lastPing = new Date();
- this.emitAll(EVENT_PING);
- }
-
- private void onpong() {
- this.emitAll(EVENT_PONG,
- null != this.lastPing ? new Date().getTime() - this.lastPing.getTime() : 0);
- }
-
- private void ondata(String data) {
- this.decoder.add(data);
- }
-
- private void ondata(byte[] data) {
- this.decoder.add(data);
- }
-
private void ondecoded(Packet packet) {
this.emit(EVENT_PACKET, packet);
}
private void onerror(Exception err) {
logger.log(Level.FINE, "error", err);
- this.emitAll(EVENT_ERROR, err);
+ this.emit(EVENT_ERROR, err);
}
/**
@@ -444,41 +371,31 @@ private void onerror(Exception err) {
* @return a socket instance for the namespace.
*/
public Socket socket(final String nsp, Options opts) {
- Socket socket = this.nsps.get(nsp);
- if (socket == null) {
- socket = new Socket(this, nsp, opts);
- Socket _socket = this.nsps.putIfAbsent(nsp, socket);
- if (_socket != null) {
- socket = _socket;
- } else {
- final Manager self = this;
- final Socket s = socket;
- socket.on(Socket.EVENT_CONNECTING, new Listener() {
- @Override
- public void call(Object... args) {
- self.connecting.add(s);
- }
- });
- socket.on(Socket.EVENT_CONNECT, new Listener() {
- @Override
- public void call(Object... objects) {
- s.id = self.generateId(nsp);
- }
- });
+ synchronized (this.nsps) {
+ Socket socket = this.nsps.get(nsp);
+ if (socket == null) {
+ socket = new Socket(this, nsp, opts);
+ this.nsps.put(nsp, socket);
}
+ return socket;
}
- return socket;
}
public Socket socket(String nsp) {
return socket(nsp, null);
}
- /*package*/ void destroy(Socket socket) {
- this.connecting.remove(socket);
- if (!this.connecting.isEmpty()) return;
+ /*package*/ void destroy() {
+ synchronized (this.nsps) {
+ for (Socket socket : this.nsps.values()) {
+ if (socket.isActive()) {
+ logger.fine("socket is still active, skipping close");
+ return;
+ }
+ }
- this.close();
+ this.close();
+ }
}
/*package*/ void packet(Packet packet) {
@@ -487,10 +404,6 @@ public Socket socket(String nsp) {
}
final Manager self = this;
- if (packet.query != null && !packet.query.isEmpty() && packet.type == Parser.CONNECT) {
- packet.nsp += "?" + packet.query;
- }
-
if (!self.encoding) {
self.encoding = true;
this.encoder.encode(packet, new Parser.Encoder.Callback() {
@@ -528,7 +441,6 @@ private void cleanup() {
this.packetBuffer.clear();
this.encoding = false;
- this.lastPing = null;
this.decoder.destroy();
}
@@ -569,7 +481,7 @@ private void reconnect() {
if (this.backoff.getAttempts() >= this._reconnectionAttempts) {
logger.fine("reconnect failed");
this.backoff.reset();
- this.emitAll(EVENT_RECONNECT_FAILED);
+ this.emit(EVENT_RECONNECT_FAILED);
this.reconnecting = false;
} else {
long delay = this.backoff.duration();
@@ -587,8 +499,7 @@ public void run() {
logger.fine("attempting reconnect");
int attempts = self.backoff.getAttempts();
- self.emitAll(EVENT_RECONNECT_ATTEMPT, attempts);
- self.emitAll(EVENT_RECONNECTING, attempts);
+ self.emit(EVENT_RECONNECT_ATTEMPT, attempts);
// check again for the case socket closed in above events
if (self.skipReconnect) return;
@@ -600,7 +511,7 @@ public void call(Exception err) {
logger.fine("reconnect attempt error");
self.reconnecting = false;
self.reconnect();
- self.emitAll(EVENT_RECONNECT_ERROR, err);
+ self.emit(EVENT_RECONNECT_ERROR, err);
} else {
logger.fine("reconnect success");
self.onreconnect();
@@ -625,14 +536,13 @@ private void onreconnect() {
int attempts = this.backoff.getAttempts();
this.reconnecting = false;
this.backoff.reset();
- this.updateSocketIds();
- this.emitAll(EVENT_RECONNECT, attempts);
+ this.emit(EVENT_RECONNECT, attempts);
}
- public static interface OpenCallback {
+ public interface OpenCallback {
- public void call(Exception err);
+ void call(Exception err);
}
@@ -652,6 +562,7 @@ public static class Options extends io.socket.engineio.client.Socket.Options {
public double randomizationFactor;
public Parser.Encoder encoder;
public Parser.Decoder decoder;
+ public Map auth;
/**
* Connection timeout (ms). Set -1 to disable.
diff --git a/src/main/java/io/socket/client/On.java b/src/main/java/io/socket/client/On.java
index b962f131..26b46f34 100644
--- a/src/main/java/io/socket/client/On.java
+++ b/src/main/java/io/socket/client/On.java
@@ -16,8 +16,8 @@ public void destroy() {
};
}
- public static interface Handle {
+ public interface Handle {
- public void destroy();
+ void destroy();
}
}
diff --git a/src/main/java/io/socket/client/Socket.java b/src/main/java/io/socket/client/Socket.java
index 369d6244..2227a30d 100644
--- a/src/main/java/io/socket/client/Socket.java
+++ b/src/main/java/io/socket/client/Socket.java
@@ -8,13 +8,9 @@
import org.json.JSONException;
import org.json.JSONObject;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -30,8 +26,6 @@ public class Socket extends Emitter {
*/
public static final String EVENT_CONNECT = "connect";
- public static final String EVENT_CONNECTING = "connecting";
-
/**
* Called on a disconnection.
*/
@@ -45,62 +39,39 @@ public class Socket extends Emitter {
* (Exception) error data.
*
*/
- public static final String EVENT_ERROR = "error";
-
- public static final String EVENT_MESSAGE = "message";
-
- public static final String EVENT_CONNECT_ERROR = Manager.EVENT_CONNECT_ERROR;
-
- public static final String EVENT_CONNECT_TIMEOUT = Manager.EVENT_CONNECT_TIMEOUT;
-
- public static final String EVENT_RECONNECT = Manager.EVENT_RECONNECT;
-
- public static final String EVENT_RECONNECT_ERROR = Manager.EVENT_RECONNECT_ERROR;
-
- public static final String EVENT_RECONNECT_FAILED = Manager.EVENT_RECONNECT_FAILED;
-
- public static final String EVENT_RECONNECT_ATTEMPT = Manager.EVENT_RECONNECT_ATTEMPT;
-
- public static final String EVENT_RECONNECTING = Manager.EVENT_RECONNECTING;
-
- public static final String EVENT_PING = Manager.EVENT_PING;
+ public static final String EVENT_CONNECT_ERROR = "connect_error";
- public static final String EVENT_PONG = Manager.EVENT_PONG;
+ static final String EVENT_MESSAGE = "message";
- protected static Map events = new HashMap() {{
+ protected static Map RESERVED_EVENTS = new HashMap() {{
put(EVENT_CONNECT, 1);
put(EVENT_CONNECT_ERROR, 1);
- put(EVENT_CONNECT_TIMEOUT, 1);
- put(EVENT_CONNECTING, 1);
put(EVENT_DISCONNECT, 1);
- put(EVENT_ERROR, 1);
- put(EVENT_RECONNECT, 1);
- put(EVENT_RECONNECT_ATTEMPT, 1);
- put(EVENT_RECONNECT_FAILED, 1);
- put(EVENT_RECONNECT_ERROR, 1);
- put(EVENT_RECONNECTING, 1);
- put(EVENT_PING, 1);
- put(EVENT_PONG, 1);
+ // used on the server-side
+ put("disconnecting", 1);
+ put("newListener", 1);
+ put("removeListener", 1);
}};
/*package*/ String id;
private volatile boolean connected;
private int ids;
- private String nsp;
- private Manager io;
- private String query;
- private Map acks = new HashMap();
+ private final String nsp;
+ private final Manager io;
+ private final Map auth;
+ private final Map acks = new ConcurrentHashMap<>();
private Queue subs;
- private final Queue> receiveBuffer = new LinkedList>();
- private final Queue> sendBuffer = new LinkedList>();
+ private final Queue> receiveBuffer = new ConcurrentLinkedQueue<>();
+ private final Queue> sendBuffer = new ConcurrentLinkedQueue<>();
+
+ private final ConcurrentLinkedQueue onAnyIncomingListeners = new ConcurrentLinkedQueue<>();
+ private final ConcurrentLinkedQueue onAnyOutgoingListeners = new ConcurrentLinkedQueue<>();
public Socket(Manager io, String nsp, Manager.Options opts) {
this.io = io;
this.nsp = nsp;
- if (opts != null) {
- this.query = opts.query;
- }
+ this.auth = opts != null ? opts.auth : null;
}
private void subEvents() {
@@ -120,6 +91,14 @@ public void call(Object... args) {
Socket.this.onpacket((Packet) args[0]);
}
}));
+ add(On.on(io, Manager.EVENT_ERROR, new Listener() {
+ @Override
+ public void call(Object... args) {
+ if (!Socket.this.connected) {
+ Socket.super.emit(EVENT_CONNECT_ERROR, args[0]);
+ }
+ }
+ }));
add(On.on(io, Manager.EVENT_CLOSE, new Listener() {
@Override
public void call(Object... args) {
@@ -129,6 +108,10 @@ public void call(Object... args) {
}};
}
+ public boolean isActive() {
+ return this.subs != null;
+ }
+
/**
* Connects the socket.
*/
@@ -141,7 +124,6 @@ public void run() {
Socket.this.subEvents();
Socket.this.io.open(); // ensure open
if (Manager.ReadyState.OPEN == Socket.this.io.readyState) Socket.this.onopen();
- Socket.this.emit(EVENT_CONNECTING);
}
});
return this;
@@ -179,14 +161,13 @@ public void run() {
*/
@Override
public Emitter emit(final String event, final Object... args) {
+ if (RESERVED_EVENTS.containsKey(event)) {
+ throw new RuntimeException("'" + event + "' is a reserved event name");
+ }
+
EventThread.exec(new Runnable() {
@Override
public void run() {
- if (events.containsKey(event)) {
- Socket.super.emit(event, args);
- return;
- }
-
Ack ack;
Object[] _args;
int lastIndex = args.length - 1;
@@ -229,11 +210,35 @@ public void run() {
}
}
- Packet packet = new Packet(Parser.EVENT, jsonArgs);
+ Packet packet = new Packet<>(Parser.EVENT, jsonArgs);
if (ack != null) {
- logger.fine(String.format("emitting packet with ack id %d", ids));
- Socket.this.acks.put(ids, ack);
+ final int ackId = Socket.this.ids;
+
+ logger.fine(String.format("emitting packet with ack id %d", ackId));
+
+ if (ack instanceof AckWithTimeout) {
+ final AckWithTimeout ackWithTimeout = (AckWithTimeout) ack;
+ ackWithTimeout.schedule(new TimerTask() {
+ @Override
+ public void run() {
+ // remove the ack from the map (to prevent an actual acknowledgement)
+ acks.remove(ackId);
+
+ // remove the packet from the buffer (if applicable)
+ Iterator> iterator = sendBuffer.iterator();
+ while (iterator.hasNext()) {
+ if (iterator.next().id == ackId) {
+ iterator.remove();
+ }
+ }
+
+ ackWithTimeout.onTimeout();
+ }
+ });
+ }
+
+ Socket.this.acks.put(ackId, ack);
packet.id = ids++;
}
@@ -248,6 +253,14 @@ public void run() {
}
private void packet(Packet packet) {
+ if (packet.type == Parser.EVENT) {
+ if (!onAnyOutgoingListeners.isEmpty()) {
+ Object[] argsAsArray = toArray((JSONArray) packet.data);
+ for (Listener listener : onAnyOutgoingListeners) {
+ listener.call(argsAsArray);
+ }
+ }
+ }
packet.nsp = this.nsp;
this.io.packet(packet);
}
@@ -255,14 +268,10 @@ private void packet(Packet packet) {
private void onopen() {
logger.fine("transport is open - connecting");
- if (!"/".equals(this.nsp)) {
- if (this.query != null && !this.query.isEmpty()) {
- Packet packet = new Packet(Parser.CONNECT);
- packet.query = this.query;
- this.packet(packet);
- } else {
- this.packet(new Packet(Parser.CONNECT));
- }
+ if (this.auth != null) {
+ this.packet(new Packet<>(Parser.CONNECT, new JSONObject(this.auth)));
+ } else {
+ this.packet(new Packet<>(Parser.CONNECT));
}
}
@@ -272,24 +281,41 @@ private void onclose(String reason) {
}
this.connected = false;
this.id = null;
- this.emit(EVENT_DISCONNECT, reason);
+ super.emit(EVENT_DISCONNECT, reason);
+ this.clearAcks();
+ }
+
+ /**
+ * Clears the acknowledgement handlers upon disconnection, since the client will never receive an acknowledgement from
+ * the server.
+ */
+ private void clearAcks() {
+ for (Ack ack : this.acks.values()) {
+ if (ack instanceof AckWithTimeout) {
+ ((AckWithTimeout) ack).onTimeout();
+ }
+ // note: basic Ack objects have no way to report an error, so they are simply ignored here
+ }
+ this.acks.clear();
}
private void onpacket(Packet packet) {
if (!this.nsp.equals(packet.nsp)) return;
switch (packet.type) {
- case Parser.CONNECT:
- this.onconnect();
- break;
-
- case Parser.EVENT: {
- @SuppressWarnings("unchecked")
- Packet p = (Packet) packet;
- this.onevent(p);
+ case Parser.CONNECT: {
+ if (packet.data instanceof JSONObject && ((JSONObject) packet.data).has("sid")) {
+ try {
+ this.onconnect(((JSONObject) packet.data).getString("sid"));
+ return;
+ } catch (JSONException e) {}
+ } else {
+ super.emit(EVENT_CONNECT_ERROR, new SocketIOException("It seems you are trying to reach a Socket.IO server in v2.x with a v3.x client, which is not possible"));
+ }
break;
}
+ case Parser.EVENT:
case Parser.BINARY_EVENT: {
@SuppressWarnings("unchecked")
Packet p = (Packet) packet;
@@ -297,13 +323,7 @@ private void onpacket(Packet packet) {
break;
}
- case Parser.ACK: {
- @SuppressWarnings("unchecked")
- Packet p = (Packet) packet;
- this.onack(p);
- break;
- }
-
+ case Parser.ACK:
case Parser.BINARY_ACK: {
@SuppressWarnings("unchecked")
Packet p = (Packet) packet;
@@ -315,14 +335,15 @@ private void onpacket(Packet packet) {
this.ondisconnect();
break;
- case Parser.ERROR:
- this.emit(EVENT_ERROR, packet.data);
+ case Parser.CONNECT_ERROR:
+ this.destroy();
+ super.emit(EVENT_CONNECT_ERROR, packet.data);
break;
}
}
private void onevent(Packet packet) {
- List