h4ckx0r/socket.io-client-java
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add support for Socket.IO v3

Browse Source 
Including:

- 969debe88c
- 6494f61be0
- 132f8ec918
- f8f60fc860

Reference: https://github.com/socketio/socket.io-protocol#difference-between-v5-and-v4
This commit is contained in:
Damien Arrachequesne
2020-12-14 15:30:23 +01:00
parent 48bf83f34c
commit 79cb27fc97
15 changed files with 276 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/main/java/io/socket/client/IO.java
View File

@@ -72,6 +72,11 @@ public class IO {
boolean newConnection = opts.forceNew || !opts.multiplex || sameNamespace;
Manager io;
String query = parsed.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,11 +92,6 @@ public class IO {
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);
}

149
src/main/java/io/socket/client/Manager.java
View File

@@ -2,6 +2,7 @@ package io.socket.client;
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.Call;
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)
*/
@@ -98,8 +74,6 @@ public class Manager extends Emitter {
private double _randomizationFactor;
private Backoff backoff;
private long _timeout;
private Set<Socket> connecting = new HashSet<Socket>();
private Date lastPing;
private URI uri;
private List<Packet> packetBuffer;
private Queue<On.Handle> subs;
@@ -160,28 +134,6 @@ public class Manager extends Emitter {
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<String, Socket> 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 +259,7 @@ public class Manager extends Emitter {
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);
@@ -334,7 +286,6 @@ public class Manager extends Emitter {
openSub.destroy();
socket.close();
socket.emit(Engine.EVENT_ERROR, new SocketIOException("timeout"));
self.emitAll(EVENT_CONNECT_TIMEOUT, timeout);
}
});
}
@@ -377,18 +328,6 @@ public class Manager extends Emitter {
}
}
}));
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,22 +348,20 @@ public class Manager extends Emitter {
});
}
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);
try {
this.decoder.add(data);
} catch (DecodingException e) {
this.onerror(e);
}
}
private void ondata(byte[] data) {
this.decoder.add(data);
try {
this.decoder.add(data);
} catch (DecodingException e) {
this.onerror(e);
}
}
private void ondecoded(Packet packet) {
@@ -433,7 +370,7 @@ public class Manager extends Emitter {
private void onerror(Exception err) {
logger.log(Level.FINE, "error", err);
this.emitAll(EVENT_ERROR, err);
this.emit(EVENT_ERROR, err);
}
/**
@@ -444,41 +381,31 @@ public class Manager extends Emitter {
* @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 +414,6 @@ public class Manager extends Emitter {
}
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 +451,6 @@ public class Manager extends Emitter {
this.packetBuffer.clear();
this.encoding = false;
this.lastPing = null;
this.decoder.destroy();
}
@@ -569,7 +491,7 @@ public class Manager extends Emitter {
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 +509,7 @@ public class Manager extends Emitter {
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 +521,7 @@ public class Manager extends Emitter {
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,8 +546,7 @@ public class Manager extends Emitter {
int attempts = this.backoff.getAttempts();
this.reconnecting = false;
this.backoff.reset();
this.updateSocketIds();
this.emitAll(EVENT_RECONNECT, attempts);
this.emit(EVENT_RECONNECT, attempts);
}
@@ -652,6 +572,7 @@ public class Manager extends Emitter {
public double randomizationFactor;
public Parser.Encoder encoder;
public Parser.Decoder decoder;
public Map<String, String> auth;
/**
* Connection timeout (ms). Set -1 to disable.

109
src/main/java/io/socket/client/Socket.java
View File

@@ -8,13 +8,7 @@ import org.json.JSONArray;
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.logging.Level;
import java.util.logging.Logger;
@@ -30,8 +24,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,42 +37,18 @@ public class Socket extends Emitter {
* <li>(Exception) error data.</li>
* </ul>
*/
public static final String EVENT_ERROR = "error";
public static final String EVENT_CONNECT_ERROR = "connect_error";
public static final String EVENT_MESSAGE = "message";
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_PONG = Manager.EVENT_PONG;
protected static Map<String, Integer> events = new HashMap<String, Integer>() {{
protected static Map<String, Integer> RESERVED_EVENTS = new HashMap<String, Integer>() {{
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;
@@ -89,7 +57,7 @@ public class Socket extends Emitter {
private int ids;
private String nsp;
private Manager io;
private String query;
private Map<String, String> auth;
private Map<Integer, Ack> acks = new HashMap<Integer, Ack>();
private Queue<On.Handle> subs;
private final Queue<List<Object>> receiveBuffer = new LinkedList<List<Object>>();
@@ -99,7 +67,7 @@ public class Socket extends Emitter {
this.io = io;
this.nsp = nsp;
if (opts != null) {
this.query = opts.query;
this.auth = opts.auth;
}
}
@@ -120,6 +88,12 @@ public class Socket extends Emitter {
Socket.this.onpacket((Packet<?>) args[0]);
}
}));
add(On.on(io, Manager.EVENT_ERROR, new Listener() {
@Override
public void call(Object... args) {
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 +103,10 @@ public class Socket extends Emitter {
}};
}
public boolean isActive() {
return this.subs != null;
}
/**
* Connects the socket.
*/
@@ -141,7 +119,6 @@ public class Socket extends Emitter {
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 +156,13 @@ public class Socket extends Emitter {
*/
@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;
@@ -255,14 +231,10 @@ public class Socket extends Emitter {
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,16 +244,24 @@ public class Socket extends Emitter {
}
this.connected = false;
this.id = null;
this.emit(EVENT_DISCONNECT, reason);
super.emit(EVENT_DISCONNECT, reason);
}
private void onpacket(Packet<?> packet) {
if (!this.nsp.equals(packet.nsp)) return;
switch (packet.type) {
case Parser.CONNECT:
this.onconnect();
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: {
@SuppressWarnings("unchecked")
@@ -315,8 +295,8 @@ public class Socket extends Emitter {
this.ondisconnect();
break;
case Parser.ERROR:
this.emit(EVENT_ERROR, packet.data);
case Parser.CONNECT_ERROR:
super.emit(EVENT_CONNECT_ERROR, packet.data);
break;
}
}
@@ -384,9 +364,10 @@ public class Socket extends Emitter {
}
}
private void onconnect() {
private void onconnect(String id) {
this.connected = true;
this.emit(EVENT_CONNECT);
this.id = id;
super.emit(EVENT_CONNECT);
this.emitBuffered();
}
@@ -422,7 +403,7 @@ public class Socket extends Emitter {
this.subs = null;
}
this.io.destroy(this);
this.io.destroy();
}
/**

7
src/main/java/io/socket/parser/DecodingException.java Normal file
View File

@@ -0,0 +1,7 @@
package io.socket.parser;
public class DecodingException extends RuntimeException {
public DecodingException(String message) {
super(message);
}
}

16
src/main/java/io/socket/parser/IOParser.java
View File

@@ -14,10 +14,6 @@ final public class IOParser implements Parser {
private static final Logger logger = Logger.getLogger(IOParser.class.getName());
private static Packet<String> error() {
return new Packet<String>(ERROR, "parser error");
}
private IOParser() {}
final public static class Encoder implements Parser.Encoder {
@@ -128,10 +124,14 @@ final public class IOParser implements Parser {
Packet<Object> p = new Packet<Object>(Character.getNumericValue(str.charAt(0)));
if (p.type < 0 || p.type > types.length - 1) return error();
if (p.type < 0 || p.type > types.length - 1) {
throw new DecodingException("unknown packet type " + p.type);
}
if (BINARY_EVENT == p.type || BINARY_ACK == p.type) {
if (!str.contains("-") || length <= i + 1) return error();
if (!str.contains("-") || length <= i + 1) {
throw new DecodingException("illegal attachments");
}
StringBuilder attachments = new StringBuilder();
while (str.charAt(++i) != '-') {
attachments.append(str.charAt(i));
@@ -170,7 +170,7 @@ final public class IOParser implements Parser {
try {
p.id = Integer.parseInt(id.toString());
} catch (NumberFormatException e){
return error();
throw new DecodingException("invalid payload");
}
}
}
@@ -181,7 +181,7 @@ final public class IOParser implements Parser {
p.data = new JSONTokener(str.substring(i)).nextValue();
} catch (JSONException e) {
logger.log(Level.WARNING, "An error occured while retrieving data from JSONTokener", e);
return error();
throw new DecodingException("invalid payload");
}
}

1
src/main/java/io/socket/parser/Packet.java
View File

@@ -8,7 +8,6 @@ public class Packet<T> {
public String nsp;
public T data;
public int attachments;
public String query;
public Packet() {}

4
src/main/java/io/socket/parser/Parser.java
View File

@@ -25,7 +25,7 @@ public interface Parser {
/**
* Packet type `error`.
*/
public static final int ERROR = 4;
public static final int CONNECT_ERROR = 4;
/**
* Packet type `binary event`.
@@ -37,7 +37,7 @@ public interface Parser {
*/
public static final int BINARY_ACK = 6;
public static int protocol = 4;
public static int protocol = 5;
/**
* Packet types.