Skip to content

Commit 41f89a3

Browse files
darrachequesnedarrachequesne
committed
feat: add support for Engine.IO v4
Reference: https://github.com/socketio/engine.io-protocol#difference-between-v3-and-v4
1 parent b1b7002 commit 41f89a3

File tree

14 files changed

+827
-862
lines changed

14 files changed

+827
-862
lines changed

‎src/main/java/io/socket/engineio/client/Socket.java

Lines changed: 12 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,6 @@ public String toString() {
128128
/*package*/ LinkedList<Packet> writeBuffer = new LinkedList<Packet>();
129129
/*package*/ Transport transport;
130130
private Future pingTimeoutTimer;
131-
private Future pingIntervalTimer;
132131
private okhttp3.WebSocket.Factory webSocketFactory;
133132
private okhttp3.Call.Factory callFactory;
134133

@@ -137,7 +136,7 @@ public String toString() {
137136
private final Listener onHeartbeatAsListener = new Listener() {
138137
@Override
139138
public void call(Object... args) {
140-
Socket.this.onHeartbeat(args.length > 0 ? (Long)args[0]: 0);
139+
Socket.this.onHeartbeat();
141140
}
142141
};
143142

@@ -540,9 +539,14 @@ private void onPacket(Packet packet) {
540539
} catch (JSONException e) {
541540
this.emit(EVENT_ERROR, new EngineIOException(e));
542541
}
543-
} else if (Packet.PONG.equals(packet.type)) {
544-
this.setPing();
545-
this.emit(EVENT_PONG);
542+
} else if (Packet.PING.equals(packet.type)) {
543+
this.emit(EVENT_PING);
544+
EventThread.exec(new Runnable() {
545+
@Override
546+
public void run() {
547+
Socket.this.sendPacket(Packet.PONG, null);
548+
}
549+
});
546550
} else if (Packet.ERROR.equals(packet.type)) {
547551
EngineIOException err = new EngineIOException("server error");
548552
err.code = packet.data;
@@ -568,20 +572,18 @@ private void onHandshake(HandshakeData data) {
568572
this.onOpen();
569573
// In case open handler closes socket
570574
if (ReadyState.CLOSED == this.readyState) return;
571-
this.setPing();
575+
this.onHeartbeat();
572576

573577
this.off(EVENT_HEARTBEAT, this.onHeartbeatAsListener);
574578
this.on(EVENT_HEARTBEAT, this.onHeartbeatAsListener);
575579
}
576580

577-
private void onHeartbeat(long timeout) {
581+
private void onHeartbeat() {
578582
if (this.pingTimeoutTimer != null) {
579583
pingTimeoutTimer.cancel(false);
580584
}
581585

582-
if (timeout <= 0) {
583-
timeout = this.pingInterval + this.pingTimeout;
584-
}
586+
long timeout = this.pingInterval + this.pingTimeout;
585587

586588
final Socket self = this;
587589
this.pingTimeoutTimer = this.getHeartbeatScheduler().schedule(new Runnable() {
@@ -598,46 +600,6 @@ public void run() {
598600
}, timeout, TimeUnit.MILLISECONDS);
599601
}
600602

601-
private void setPing() {
602-
if (this.pingIntervalTimer != null) {
603-
pingIntervalTimer.cancel(false);
604-
}
605-
606-
final Socket self = this;
607-
this.pingIntervalTimer = this.getHeartbeatScheduler().schedule(new Runnable() {
608-
@Override
609-
public void run() {
610-
EventThread.exec(new Runnable() {
611-
@Override
612-
public void run() {
613-
if (logger.isLoggable(Level.FINE)) {
614-
logger.fine(String.format("writing ping packet - expecting pong within %sms", self.pingTimeout));
615-
}
616-
self.ping();
617-
self.onHeartbeat(self.pingTimeout);
618-
}
619-
});
620-
}
621-
}, this.pingInterval, TimeUnit.MILLISECONDS);
622-
}
623-
624-
/**
625-
* Sends a ping packet.
626-
*/
627-
private void ping() {
628-
EventThread.exec(new Runnable() {
629-
@Override
630-
public void run() {
631-
Socket.this.sendPacket(Packet.PING, new Runnable() {
632-
@Override
633-
public void run() {
634-
Socket.this.emit(EVENT_PING);
635-
}
636-
});
637-
}
638-
});
639-
}
640-
641603
private void onDrain() {
642604
for (int i = 0; i < this.prevBufferLen; i++) {
643605
this.writeBuffer.poll();
@@ -833,9 +795,6 @@ private void onClose(String reason, Exception desc) {
833795
final Socket self = this;
834796

835797
// clear timers
836-
if (this.pingIntervalTimer != null) {
837-
this.pingIntervalTimer.cancel(false);
838-
}
839798
if (this.pingTimeoutTimer != null) {
840799
this.pingTimeoutTimer.cancel(false);
841800
}

‎src/main/java/io/socket/engineio/client/Transport.java

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
import io.socket.engineio.parser.Packet;
88
import io.socket.engineio.parser.Parser;
99
import io.socket.thread.EventThread;
10-
import io.socket.utf8.UTF8Exception;
1110
import okhttp3.Call;
1211
import okhttp3.WebSocket;
1312

@@ -96,11 +95,7 @@ public void send(final Packet[] packets) {
9695
@Override
9796
public void run() {
9897
if (Transport.this.readyState == ReadyState.OPEN) {
99-
try {
100-
Transport.this.write(packets);
101-
} catch (UTF8Exception err) {
102-
throw new RuntimeException(err);
103-
}
98+
Transport.this.write(packets);
10499
} else {
105100
throw new RuntimeException("Transport not open");
106101
}
@@ -131,7 +126,7 @@ protected void onClose() {
131126
this.emit(EVENT_CLOSE);
132127
}
133128

134-
abstract protected void write(Packet[] packets) throws UTF8Exception;
129+
abstract protected void write(Packet[] packets);
135130

136131
abstract protected void doOpen();
137132

‎src/main/java/io/socket/engineio/client/transports/Polling.java

Lines changed: 6 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
import io.socket.engineio.parser.Parser;
88
import io.socket.parseqs.ParseQS;
99
import io.socket.thread.EventThread;
10-
import io.socket.utf8.UTF8Exception;
1110
import io.socket.yeast.Yeast;
1211

1312
import java.util.HashMap;
@@ -129,13 +128,7 @@ public boolean call(Packet packet, int index, int total) {
129128
}
130129
};
131130

132-
if (data instanceof String) {
133-
@SuppressWarnings("unchecked")
134-
Parser.DecodePayloadCallback<String> tempCallback = callback;
135-
Parser.decodePayload((String)data, tempCallback);
136-
} else if (data instanceof byte[]) {
137-
Parser.decodePayload((byte[])data, callback);
138-
}
131+
Parser.decodePayload((String) data, callback);
139132

140133
if (this.readyState != ReadyState.CLOSED) {
141134
this.polling = false;
@@ -158,11 +151,7 @@ protected void doClose() {
158151
@Override
159152
public void call(Object... args) {
160153
logger.fine("writing close packet");
161-
try {
162-
self.write(new Packet[]{new Packet(Packet.CLOSE)});
163-
} catch (UTF8Exception err) {
164-
throw new RuntimeException(err);
165-
}
154+
self.write(new Packet[]{new Packet(Packet.CLOSE)});
166155
}
167156
};
168157

@@ -177,7 +166,7 @@ public void call(Object... args) {
177166
}
178167
}
179168

180-
protected void write(Packet[] packets) throws UTF8Exception {
169+
protected void write(Packet[] packets) {
181170
final Polling self = this;
182171
this.writable = false;
183172
final Runnable callbackfn = new Runnable() {
@@ -188,16 +177,10 @@ public void run() {
188177
}
189178
};
190179

191-
Parser.encodePayload(packets, new Parser.EncodeCallback() {
180+
Parser.encodePayload(packets, new Parser.EncodeCallback<String>() {
192181
@Override
193-
public void call(Object data) {
194-
if (data instanceof byte[]) {
195-
self.doWrite((byte[])data, callbackfn);
196-
} else if (data instanceof String) {
197-
self.doWrite((String)data, callbackfn);
198-
} else {
199-
logger.warning("Unexpected data: " + data);
200-
}
182+
public void call(String data) {
183+
self.doWrite(data, callbackfn);
201184
}
202185
});
203186
}
@@ -229,8 +212,6 @@ protected String uri() {
229212
return schema + "://" + (ipv6 ? "[" + this.hostname + "]" : this.hostname) + port + this.path + derivedQuery;
230213
}
231214

232-
abstract protected void doWrite(byte[] data, Runnable fn);
233-
234215
abstract protected void doWrite(String data, Runnable fn);
235216

236217
abstract protected void doPoll();

‎src/main/java/io/socket/engineio/client/transports/PollingXHR.java

Lines changed: 8 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -67,17 +67,8 @@ public void run() {
6767
return req;
6868
}
6969

70-
@Override
71-
protected void doWrite(byte[] data, final Runnable fn) {
72-
this.doWrite((Object) data, fn);
73-
}
74-
7570
@Override
7671
protected void doWrite(String data, final Runnable fn) {
77-
this.doWrite((Object) data, fn);
78-
}
79-
80-
private void doWrite(Object data, final Runnable fn) {
8172
Request.Options opts = new Request.Options();
8273
opts.method = "POST";
8374
opts.data = data;
@@ -121,11 +112,7 @@ public void call(final Object... args) {
121112
@Override
122113
public void run() {
123114
Object arg = args.length > 0 ? args[0] : null;
124-
if (arg instanceof String) {
125-
self.onData((String)arg);
126-
} else if (arg instanceof byte[]) {
127-
self.onData((byte[])arg);
128-
}
115+
self.onData((String)arg);
129116
}
130117
});
131118
}
@@ -153,16 +140,14 @@ public static class Request extends Emitter {
153140
public static final String EVENT_REQUEST_HEADERS = "requestHeaders";
154141
public static final String EVENT_RESPONSE_HEADERS = "responseHeaders";
155142

156-
private static final String BINARY_CONTENT_TYPE = "application/octet-stream";
157143
private static final String TEXT_CONTENT_TYPE = "text/plain;charset=UTF-8";
158144

159-
private static final MediaType BINARY_MEDIA_TYPE = MediaType.parse(BINARY_CONTENT_TYPE);
160145
private static final MediaType TEXT_MEDIA_TYPE = MediaType.parse(TEXT_CONTENT_TYPE);
161146

162147
private String method;
163148
private String uri;
164149

165-
private Object data;
150+
private String data;
166151

167152
private Call.Factory callFactory;
168153
private Response response;
@@ -181,20 +166,15 @@ public void create() {
181166
Map<String, List<String>> headers = new TreeMap<String, List<String>>(String.CASE_INSENSITIVE_ORDER);
182167

183168
if ("POST".equals(this.method)) {
184-
if (this.data instanceof byte[]) {
185-
headers.put("Content-type", new LinkedList<String>(Collections.singletonList(BINARY_CONTENT_TYPE)));
186-
} else {
187-
headers.put("Content-type", new LinkedList<String>(Collections.singletonList(TEXT_CONTENT_TYPE)));
188-
}
169+
headers.put("Content-type", new LinkedList<String>(Collections.singletonList(TEXT_CONTENT_TYPE)));
189170
}
190171

191172
headers.put("Accept", new LinkedList<String>(Collections.singletonList("*/*")));
192173

193174
this.onRequestHeaders(headers);
194175

195176
if (LOGGABLE_FINE) {
196-
logger.fine(String.format("sending xhr with url %s | data %s", this.uri,
197-
this.data instanceof byte[] ? Arrays.toString((byte[]) this.data) : this.data));
177+
logger.fine(String.format("sending xhr with url %s | data %s", this.uri, this.data));
198178
}
199179

200180
okhttp3.Request.Builder requestBuilder = new okhttp3.Request.Builder();
@@ -204,10 +184,8 @@ public void create() {
204184
}
205185
}
206186
RequestBody body = null;
207-
if (this.data instanceof byte[]) {
208-
body = RequestBody.create(BINARY_MEDIA_TYPE, (byte[])this.data);
209-
} else if (this.data instanceof String) {
210-
body = RequestBody.create(TEXT_MEDIA_TYPE, (String)this.data);
187+
if (this.data != null) {
188+
body = RequestBody.create(TEXT_MEDIA_TYPE, this.data);
211189
}
212190

213191
okhttp3.Request request = requestBuilder
@@ -249,11 +227,6 @@ private void onData(String data) {
249227
this.onSuccess();
250228
}
251229

252-
private void onData(byte[] data) {
253-
this.emit(EVENT_DATA, data);
254-
this.onSuccess();
255-
}
256-
257230
private void onError(Exception err) {
258231
this.emit(EVENT_ERROR, err);
259232
}
@@ -268,14 +241,9 @@ private void onResponseHeaders(Map<String, List<String>> headers) {
268241

269242
private void onLoad() {
270243
ResponseBody body = response.body();
271-
MediaType mediaType = body.contentType();
272244

273245
try {
274-
if (mediaType != null && BINARY_CONTENT_TYPE.equalsIgnoreCase(mediaType.toString())) {
275-
this.onData(body.bytes());
276-
} else {
277-
this.onData(body.string());
278-
}
246+
this.onData(body.string());
279247
} catch (IOException e) {
280248
this.onError(e);
281249
}
@@ -285,7 +253,7 @@ public static class Options {
285253

286254
public String uri;
287255
public String method;
288-
public Object data;
256+
public String data;
289257
public Call.Factory callFactory;
290258
}
291259
}

‎src/main/java/io/socket/engineio/client/transports/WebSocket.java

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,25 +1,24 @@
11
package io.socket.engineio.client.transports;
22

33

4-
import java.util.HashMap;
5-
import java.util.List;
6-
import java.util.Map;
7-
import java.util.TreeMap;
8-
import java.util.logging.Logger;
9-
104
import io.socket.engineio.client.Transport;
115
import io.socket.engineio.parser.Packet;
126
import io.socket.engineio.parser.Parser;
137
import io.socket.parseqs.ParseQS;
148
import io.socket.thread.EventThread;
15-
import io.socket.utf8.UTF8Exception;
169
import io.socket.yeast.Yeast;
1710
import okhttp3.OkHttpClient;
1811
import okhttp3.Request;
1912
import okhttp3.Response;
2013
import okhttp3.WebSocketListener;
2114
import okio.ByteString;
2215

16+
import java.util.HashMap;
17+
import java.util.List;
18+
import java.util.Map;
19+
import java.util.TreeMap;
20+
import java.util.logging.Logger;
21+
2322

2423
public class WebSocket extends Transport {
2524

@@ -111,7 +110,7 @@ public void run() {
111110
});
112111
}
113112

114-
protected void write(Packet[] packets) throws UTF8Exception {
113+
protected void write(Packet[] packets) {
115114
final WebSocket self = this;
116115
this.writable = false;
117116

0 commit comments

Comments
 (0)