Skip to content

Commit

Permalink
feat: add support for Socket.IO v3
Browse files Browse the repository at this point in the history 
Including:

- socketio/socket.io-client@969debe
- socketio/socket.io-client@6494f61
- socketio/socket.io-client@132f8ec
- socketio/socket.io-client@f8f60fc

Reference: https://github.com/socketio/socket.io-protocol#difference-between-v5-and-v4
  • Loading branch information
@darrachequesne
darrachequesne committed Dec 14, 2020
1 parent 48bf83f commit 79cb27f
Show file tree
Hide file tree
Showing 15 changed files with 276 additions and 549 deletions.
4 changes: 2 additions & 2 deletions pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>io.socket</groupId>
<artifactId>socket.io-client</artifactId>
<version>1.0.2-SNAPSHOT</version>
<version>2.0.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>socket.io-client</name>
<description>Socket.IO Client Library for Java</description>
Expand Down Expand Up @@ -62,7 +62,7 @@
<dependency>
<groupId>io.socket</groupId>
<artifactId>engine.io-client</artifactId>
<version>1.0.1</version>
<version>2.0.0</version>
</dependency>
<dependency>
<groupId>org.json</groupId>
Expand Down
10 changes: 5 additions & 5 deletions src/main/java/io/socket/client/IO.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,11 @@ public static Socket socket(URI uri, Options opts) {
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));
Expand All @@ -87,11 +92,6 @@ 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);
}

Expand Down
149 changes: 35 additions & 114 deletions src/main/java/io/socket/client/Manager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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.
*/
Expand All @@ -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)
*/
Expand All @@ -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;
Expand Down Expand Up @@ -160,28 +134,6 @@ public Manager(URI uri, Options opts) {
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;
}
Expand Down Expand Up @@ -307,7 +259,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);
Expand All @@ -334,7 +286,6 @@ public void run() {
openSub.destroy();
socket.close();
socket.emit(Engine.EVENT_ERROR, new SocketIOException("timeout"));
self.emitAll(EVENT_CONNECT_TIMEOUT, timeout);
}
});
}
Expand Down Expand Up @@ -377,18 +328,6 @@ public void call(Object... objects) {
}
}
}));
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) {
Expand All @@ -409,22 +348,20 @@ 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);
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) {
Expand All @@ -433,7 +370,7 @@ private void ondecoded(Packet packet) {

private void onerror(Exception err) {
logger.log(Level.FINE, "error", err);
this.emitAll(EVENT_ERROR, err);
this.emit(EVENT_ERROR, err);
}

/**
Expand All @@ -444,41 +381,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) {
Expand All @@ -487,10 +414,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() {
Expand Down Expand Up @@ -528,7 +451,6 @@ private void cleanup() {

this.packetBuffer.clear();
this.encoding = false;
this.lastPing = null;

this.decoder.destroy();
}
Expand Down Expand Up @@ -569,7 +491,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();
Expand All @@ -587,8 +509,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;
Expand All @@ -600,7 +521,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();
Expand All @@ -625,8 +546,7 @@ 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);
}


Expand All @@ -652,6 +572,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<String, String> auth;

/**
* Connection timeout (ms). Set -1 to disable.
Expand Down
Loading

0 comments on commit 79cb27f

Please sign in to comment.