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

move package name

Browse Source
This commit is contained in:
nkzawa
2015-08-31 02:34:28 +09:00
parent 411001a04f
commit acddf2f69c
35 changed files with 110 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
src/main/java/io/socket/emitter/Emitter.java Normal file
View File

@@ -0,0 +1,167 @@
package io.socket.emitter;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
/**
* The event emitter which is ported from the JavaScript module. This class is thread-safe.
*
* @see <a href="https://github.com/component/emitter">https://github.com/component/emitter</a>
*/
public class Emitter {
private ConcurrentMap<String, ConcurrentLinkedQueue<Listener>> callbacks
= new ConcurrentHashMap<String, ConcurrentLinkedQueue<Listener>>();
/**
* Listens on the event.
* @param event event name.
* @param fn
* @return a reference to this object.
*/
public Emitter on(String event, Listener fn) {
ConcurrentLinkedQueue<Listener> callbacks = this.callbacks.get(event);
if (callbacks == null) {
callbacks = new ConcurrentLinkedQueue <Listener>();
ConcurrentLinkedQueue<Listener> _callbacks = this.callbacks.putIfAbsent(event, callbacks);
if (_callbacks != null) {
callbacks = _callbacks;
}
}
callbacks.add(fn);
return this;
}
/**
* Adds a one time listener for the event.
*
* @param event an event name.
* @param fn
* @return a reference to this object.
*/
public Emitter once(final String event, final Listener fn) {
this.on(event, new OnceListener(event, fn));
return this;
}
/**
* Removes all registered listeners.
*
* @return a reference to this object.
*/
public Emitter off() {
this.callbacks.clear();
return this;
}
/**
* Removes all listeners of the specified event.
*
* @param event an event name.
* @return a reference to this object.
*/
public Emitter off(String event) {
this.callbacks.remove(event);
return this;
}
/**
* Removes the listener.
*
* @param event an event name.
* @param fn
* @return a reference to this object.
*/
public Emitter off(String event, Listener fn) {
ConcurrentLinkedQueue<Listener> callbacks = this.callbacks.get(event);
if (callbacks != null) {
Iterator<Listener> it = callbacks.iterator();
while (it.hasNext()) {
Listener internal = it.next();
if (Emitter.sameAs(fn, internal)) {
it.remove();
break;
}
}
}
return this;
}
private static boolean sameAs(Listener fn, Listener internal) {
if (fn.equals(internal)) {
return true;
} else if (internal instanceof OnceListener) {
return fn.equals(((OnceListener) internal).fn);
} else {
return false;
}
}
/**
* Executes each of listeners with the given args.
*
* @param event an event name.
* @param args
* @return a reference to this object.
*/
public Emitter emit(String event, Object... args) {
ConcurrentLinkedQueue<Listener> callbacks = this.callbacks.get(event);
if (callbacks != null) {
for (Listener fn : callbacks) {
fn.call(args);
}
}
return this;
}
/**
* Returns a list of listeners for the specified event.
*
* @param event an event name.
* @return a reference to this object.
*/
public List<Listener> listeners(String event) {
ConcurrentLinkedQueue<Listener> callbacks = this.callbacks.get(event);
return callbacks != null ?
new ArrayList<Listener>(callbacks) : new ArrayList<Listener>(0);
}
/**
* Check if this emitter has listeners for the specified event.
*
* @param event an event name.
* @return a reference to this object.
*/
public boolean hasListeners(String event) {
ConcurrentLinkedQueue<Listener> callbacks = this.callbacks.get(event);
return callbacks != null && !callbacks.isEmpty();
}
public static interface Listener {
public void call(Object... args);
}
private class OnceListener implements Listener {
public final String event;
public final Listener fn;
public OnceListener(String event, Listener fn) {
this.event = event;
this.fn = fn;
}
@Override
public void call(Object... args) {
Emitter.this.off(this.event, this);
this.fn.call(args);
}
}
}

23
src/main/java/io/socket/engineio/client/EngineIOException.java Normal file
View File

@@ -0,0 +1,23 @@
package io.socket.engineio.client;
public class EngineIOException extends Exception {
public String transport;
public Object code;
public EngineIOException() {
super();
}
public EngineIOException(String message) {
super(message);
}
public EngineIOException(String message, Throwable cause) {
super(message, cause);
}
public EngineIOException(Throwable cause) {
super(cause);
}
}

32
src/main/java/io/socket/engineio/client/HandshakeData.java Normal file
View File

@@ -0,0 +1,32 @@
package io.socket.engineio.client;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
public class HandshakeData {
public String sid;
public String[] upgrades;
public long pingInterval;
public long pingTimeout;
/*package*/ HandshakeData(String data) throws JSONException {
this(new JSONObject(data));
}
/*package*/ HandshakeData(JSONObject data) throws JSONException {
JSONArray upgrades = data.getJSONArray("upgrades");
int length = upgrades.length();
String[] _upgrades = new String[length];
for (int i = 0; i < length; i ++) {
_upgrades[i] = upgrades.getString(i);
}
this.sid = data.getString("sid");
this.upgrades = _upgrades;
this.pingInterval = data.getLong("pingInterval");
this.pingTimeout = data.getLong("pingTimeout");
}
}

876
src/main/java/io/socket/engineio/client/Socket.java Normal file
View File

@@ -0,0 +1,876 @@
package io.socket.engineio.client;
import io.socket.emitter.Emitter;
import io.socket.engineio.client.transports.Polling;
import io.socket.engineio.client.transports.PollingXHR;
import io.socket.engineio.client.transports.WebSocket;
import io.socket.engineio.parser.Packet;
import io.socket.engineio.parser.Parser;
import io.socket.parseqs.ParseQS;
import io.socket.thread.EventThread;
import org.json.JSONException;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLContext;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
/**
* The socket class for Event.IO Client.
*
* @see <a href="https://github.com/LearnBoost/engine.io-client">https://github.com/LearnBoost/engine.io-client</a>
*/
public class Socket extends Emitter {
private static final Logger logger = Logger.getLogger(Socket.class.getName());
private enum ReadyState {
OPENING, OPEN, CLOSING, CLOSED;
@Override
public String toString() {
return super.toString().toLowerCase();
}
}
/**
* Called on successful connection.
*/
public static final String EVENT_OPEN = "open";
/**
* Called on disconnection.
*/
public static final String EVENT_CLOSE = "close";
/**
* Called when data is received from the server.
*/
public static final String EVENT_MESSAGE = "message";
/**
* Called when an error occurs.
*/
public static final String EVENT_ERROR = "error";
public static final String EVENT_UPGRADE_ERROR = "upgradeError";
/**
* Called on completing a buffer flush.
*/
public static final String EVENT_FLUSH = "flush";
/**
* Called after `drain` event of transport if writeBuffer is empty.
*/
public static final String EVENT_DRAIN = "drain";
public static final String EVENT_HANDSHAKE = "handshake";
public static final String EVENT_UPGRADING = "upgrading";
public static final String EVENT_UPGRADE = "upgrade";
public static final String EVENT_PACKET = "packet";
public static final String EVENT_PACKET_CREATE = "packetCreate";
public static final String EVENT_HEARTBEAT = "heartbeat";
public static final String EVENT_DATA = "data";
/**
* Called on a new transport is created.
*/
public static final String EVENT_TRANSPORT = "transport";
private static final Runnable noop = new Runnable() {
@Override
public void run() {}
};
/**
* The protocol version.
*/
public static final int protocol = Parser.protocol;
private static boolean priorWebsocketSuccess = false;
private static SSLContext defaultSSLContext;
private static HostnameVerifier defaultHostnameVerifier;
private boolean secure;
private boolean upgrade;
private boolean timestampRequests;
private boolean upgrading;
private boolean rememberUpgrade;
/*package*/ int port;
private int policyPort;
private int prevBufferLen;
private long pingInterval;
private long pingTimeout;
private String id;
private String hostname;
private String path;
private String timestampParam;
private List<String> transports;
private List<String> upgrades;
private Map<String, String> query;
/*package*/ LinkedList<Packet> writeBuffer = new LinkedList<Packet>();
private LinkedList<Runnable> callbackBuffer = new LinkedList<Runnable>();
/*package*/ Transport transport;
private Future pingTimeoutTimer;
private Future pingIntervalTimer;
private SSLContext sslContext;
private HostnameVerifier hostnameVerifier;
private ReadyState readyState;
private ScheduledExecutorService heartbeatScheduler;
public static void setDefaultSSLContext(SSLContext sslContext) {
defaultSSLContext = sslContext;
}
public static void setDefaultHostnameVerifier(HostnameVerifier hostnameVerifier) {
defaultHostnameVerifier = hostnameVerifier;
}
public Socket() {
this(new Options());
}
/**
* Creates a socket.
*
* @param uri URI to connect.
* @throws URISyntaxException
*/
public Socket(String uri) throws URISyntaxException {
this(uri, null);
}
public Socket(URI uri) {
this(uri, null);
}
/**
* Creates a socket with options.
*
* @param uri URI to connect.
* @param opts options for socket
* @throws URISyntaxException
*/
public Socket(String uri, Options opts) throws URISyntaxException {
this(uri == null ? null : new URI(uri), opts);
}
public Socket(URI uri, Options opts) {
this(uri == null ? opts : Options.fromURI(uri, opts));
}
public Socket(Options opts) {
if (opts.host != null) {
boolean ipv6uri = opts.host.indexOf(']') != -1;
String[] pieces = ipv6uri ? opts.host.split("]:") : opts.host.split(":");
boolean ipv6 = (pieces.length > 2 || opts.host.indexOf("::") == -1);
if (ipv6) {
opts.hostname = opts.host;
} else {
opts.hostname = pieces[0];
if (ipv6uri) {
opts.hostname = opts.hostname.substring(1);
}
if (pieces.length > 1) {
opts.port = Integer.parseInt(pieces[pieces.length - 1]);
} else if (opts.port == -1) {
// if no port is specified manually, use the protocol default
opts.port = this.secure ? 443 : 80;
}
}
}
this.secure = opts.secure;
this.sslContext = opts.sslContext != null ? opts.sslContext : defaultSSLContext;
this.hostname = opts.hostname != null ? opts.hostname : "localhost";
this.port = opts.port != 0 ? opts.port : (this.secure ? 443 : 80);
this.query = opts.query != null ?
ParseQS.decode(opts.query) : new HashMap<String, String>();
this.upgrade = opts.upgrade;
this.path = (opts.path != null ? opts.path : "/engine.io").replaceAll("/$", "") + "/";
this.timestampParam = opts.timestampParam != null ? opts.timestampParam : "t";
this.timestampRequests = opts.timestampRequests;
this.transports = new ArrayList<String>(Arrays.asList(opts.transports != null ?
opts.transports : new String[]{Polling.NAME, WebSocket.NAME}));
this.policyPort = opts.policyPort != 0 ? opts.policyPort : 843;
this.rememberUpgrade = opts.rememberUpgrade;
this.hostnameVerifier = opts.hostnameVerifier != null ? opts.hostnameVerifier : defaultHostnameVerifier;
}
/**
* Connects the client.
*
* @return a reference to to this object.
*/
public Socket open() {
EventThread.exec(new Runnable() {
@Override
public void run() {
String transportName;
if (Socket.this.rememberUpgrade && Socket.priorWebsocketSuccess && Socket.this.transports.contains(WebSocket.NAME)) {
transportName = WebSocket.NAME;
} else if (0 == Socket.this.transports.size()) {
// Emit error on next tick so it can be listened to
final Socket self = Socket.this;
EventThread.nextTick(new Runnable() {
@Override
public void run() {
self.emit(Socket.EVENT_ERROR, new EngineIOException("No transports available"));
}
});
return;
} else {
transportName = Socket.this.transports.get(0);
}
Socket.this.readyState = ReadyState.OPENING;
Transport transport = Socket.this.createTransport(transportName);
Socket.this.setTransport(transport);
transport.open();
}
});
return this;
}
private Transport createTransport(String name) {
logger.fine(String.format("creating transport '%s'", name));
Map<String, String> query = new HashMap<String, String>(this.query);
query.put("EIO", String.valueOf(Parser.protocol));
query.put("transport", name);
if (this.id != null) {
query.put("sid", this.id);
}
Transport.Options opts = new Transport.Options();
opts.sslContext = this.sslContext;
opts.hostname = this.hostname;
opts.port = this.port;
opts.secure = this.secure;
opts.path = this.path;
opts.query = query;
opts.timestampRequests = this.timestampRequests;
opts.timestampParam = this.timestampParam;
opts.policyPort = this.policyPort;
opts.socket = this;
opts.hostnameVerifier = this.hostnameVerifier;
Transport transport;
if (WebSocket.NAME.equals(name)) {
transport = new WebSocket(opts);
} else if (Polling.NAME.equals(name)) {
transport = new PollingXHR(opts);
} else {
throw new RuntimeException();
}
this.emit(EVENT_TRANSPORT, transport);
return transport;
}
private void setTransport(Transport transport) {
logger.fine(String.format("setting transport %s", transport.name));
final Socket self = this;
if (this.transport != null) {
logger.fine(String.format("clearing existing transport %s", this.transport.name));
this.transport.off();
}
this.transport = transport;
transport.on(Transport.EVENT_DRAIN, new Listener() {
@Override
public void call(Object... args) {
self.onDrain();
}
}).on(Transport.EVENT_PACKET, new Listener() {
@Override
public void call(Object... args) {
self.onPacket(args.length > 0 ? (Packet) args[0] : null);
}
}).on(Transport.EVENT_ERROR, new Listener() {
@Override
public void call(Object... args) {
self.onError(args.length > 0 ? (Exception) args[0] : null);
}
}).on(Transport.EVENT_CLOSE, new Listener() {
@Override
public void call(Object... args) {
self.onClose("transport close");
}
});
}
private void probe(final String name) {
logger.fine(String.format("probing transport '%s'", name));
final Transport[] transport = new Transport[] {this.createTransport(name)};
final boolean[] failed = new boolean[] {false};
final Socket self = this;
Socket.priorWebsocketSuccess = false;
final Runnable[] cleanup = new Runnable[1];
final Listener onTransportOpen = new Listener() {
@Override
public void call(Object... args) {
if (failed[0]) return;
logger.fine(String.format("probe transport '%s' opened", name));
Packet<String> packet = new Packet<String>(Packet.PING, "probe");
transport[0].send(new Packet[] {packet});
transport[0].once(Transport.EVENT_PACKET, new Listener() {
@Override
public void call(Object... args) {
if (failed[0]) return;
Packet msg = (Packet)args[0];
if (Packet.PONG.equals(msg.type) && "probe".equals(msg.data)) {
logger.fine(String.format("probe transport '%s' pong", name));
self.upgrading = true;
self.emit(EVENT_UPGRADING, transport[0]);
if (null == transport[0]) return;
Socket.priorWebsocketSuccess = WebSocket.NAME.equals(transport[0].name);
logger.fine(String.format("pausing current transport '%s'", self.transport.name));
((Polling)self.transport).pause(new Runnable() {
@Override
public void run() {
if (failed[0]) return;
if (ReadyState.CLOSED == self.readyState) return;
logger.fine("changing transport and sending upgrade packet");
cleanup[0].run();
self.setTransport(transport[0]);
Packet packet = new Packet(Packet.UPGRADE);
transport[0].send(new Packet[]{packet});
self.emit(EVENT_UPGRADE, transport[0]);
transport[0] = null;
self.upgrading = false;
self.flush();
}
});
} else {
logger.fine(String.format("probe transport '%s' failed", name));
EngineIOException err = new EngineIOException("probe error");
err.transport = transport[0].name;
self.emit(EVENT_UPGRADE_ERROR, err);
}
}
});
}
};
final Listener freezeTransport = new Listener() {
@Override
public void call(Object... args) {
if (failed[0]) return;
failed[0] = true;
cleanup[0].run();
transport[0].close();
transport[0] = null;
}
};
// Handle any error that happens while probing
final Listener onerror = new Listener() {
@Override
public void call(Object... args) {
Object err = args[0];
EngineIOException error;
if (err instanceof Exception) {
error = new EngineIOException("probe error", (Exception)err);
} else if (err instanceof String) {
error = new EngineIOException("probe error: " + (String)err);
} else {
error = new EngineIOException("probe error");
}
error.transport = transport[0].name;
freezeTransport.call();
logger.fine(String.format("probe transport \"%s\" failed because of error: %s", name, err));
self.emit(EVENT_UPGRADE_ERROR, error);
}
};
final Listener onTransportClose = new Listener() {
@Override
public void call(Object... args) {
onerror.call("transport closed");
}
};
// When the socket is closed while we're probing
final Listener onclose = new Listener() {
@Override
public void call(Object... args) {
onerror.call("socket closed");
}
};
// When the socket is upgraded while we're probing
final Listener onupgrade = new Listener() {
@Override
public void call(Object... args) {
Transport to = (Transport)args[0];
if (transport[0] != null && !to.name.equals(transport[0].name)) {
logger.fine(String.format("'%s' works - aborting '%s'", to.name, transport[0].name));
freezeTransport.call();
}
}
};
cleanup[0] = new Runnable() {
@Override
public void run() {
transport[0].off(Transport.EVENT_OPEN, onTransportOpen);
transport[0].off(Transport.EVENT_ERROR, onerror);
transport[0].off(Transport.EVENT_CLOSE, onTransportClose);
self.off(EVENT_CLOSE, onclose);
self.off(EVENT_UPGRADING, onupgrade);
}
};
transport[0].once(Transport.EVENT_OPEN, onTransportOpen);
transport[0].once(Transport.EVENT_ERROR, onerror);
transport[0].once(Transport.EVENT_CLOSE, onTransportClose);
this.once(EVENT_CLOSE, onclose);
this.once(EVENT_UPGRADING, onupgrade);
transport[0].open();
}
private void onOpen() {
logger.fine("socket open");
this.readyState = ReadyState.OPEN;
Socket.priorWebsocketSuccess = WebSocket.NAME.equals(this.transport.name);
this.emit(EVENT_OPEN);
this.flush();
if (this.readyState == ReadyState.OPEN && this.upgrade && this.transport instanceof Polling) {
logger.fine("starting upgrade probes");
for (String upgrade: this.upgrades) {
this.probe(upgrade);
}
}
}
private void onPacket(Packet packet) {
if (this.readyState == ReadyState.OPENING || this.readyState == ReadyState.OPEN) {
logger.fine(String.format("socket received: type '%s', data '%s'", packet.type, packet.data));
this.emit(EVENT_PACKET, packet);
this.emit(EVENT_HEARTBEAT);
if (Packet.OPEN.equals(packet.type)) {
try {
this.onHandshake(new HandshakeData((String)packet.data));
} catch (JSONException e) {
this.emit(EVENT_ERROR, new EngineIOException(e));
}
} else if (Packet.PONG.equals(packet.type)) {
this.setPing();
} else if (Packet.ERROR.equals(packet.type)) {
EngineIOException err = new EngineIOException("server error");
err.code = packet.data;
this.emit(EVENT_ERROR, err);
} else if (Packet.MESSAGE.equals(packet.type)) {
this.emit(EVENT_DATA, packet.data);
this.emit(EVENT_MESSAGE, packet.data);
}
} else {
logger.fine(String.format("packet received with socket readyState '%s'", this.readyState));
}
}
private void onHandshake(HandshakeData data) {
this.emit(EVENT_HANDSHAKE, data);
this.id = data.sid;
this.transport.query.put("sid", data.sid);
this.upgrades = this.filterUpgrades(Arrays.asList(data.upgrades));
this.pingInterval = data.pingInterval;
this.pingTimeout = data.pingTimeout;
this.onOpen();
// In case open handler closes socket
if (ReadyState.CLOSED == this.readyState) return;
this.setPing();
this.off(EVENT_HEARTBEAT, this.onHeartbeatAsListener);
this.on(EVENT_HEARTBEAT, this.onHeartbeatAsListener);
}
private final Listener onHeartbeatAsListener = new Listener() {
@Override
public void call(Object... args) {
Socket.this.onHeartbeat(args.length > 0 ? (Long)args[0]: 0);
}
};
private void onHeartbeat(long timeout) {
if (this.pingTimeoutTimer != null) {
pingTimeoutTimer.cancel(false);
}
if (timeout <= 0) {
timeout = this.pingInterval + this.pingTimeout;
}
final Socket self = this;
this.pingTimeoutTimer = this.getHeartbeatScheduler().schedule(new Runnable() {
@Override
public void run() {
EventThread.exec(new Runnable() {
@Override
public void run() {
if (self.readyState == ReadyState.CLOSED) return;
self.onClose("ping timeout");
}
});
}
}, timeout, TimeUnit.MILLISECONDS);
}
private void setPing() {
if (this.pingIntervalTimer != null) {
pingIntervalTimer.cancel(false);
}
final Socket self = this;
this.pingIntervalTimer = this.getHeartbeatScheduler().schedule(new Runnable() {
@Override
public void run() {
EventThread.exec(new Runnable() {
@Override
public void run() {
logger.fine(String.format("writing ping packet - expecting pong within %sms", self.pingTimeout));
self.ping();
self.onHeartbeat(self.pingTimeout);
}
});
}
}, this.pingInterval, TimeUnit.MILLISECONDS);
}
/**
* Sends a ping packet.
*/
public void ping() {
EventThread.exec(new Runnable() {
@Override
public void run() {
Socket.this.sendPacket(Packet.PING);
}
});
}
private void onDrain() {
for (int i = 0; i < this.prevBufferLen; i++) {
Runnable callback = this.callbackBuffer.get(i);
if (callback != null) {
callback.run();
}
}
for (int i = 0; i < this.prevBufferLen; i++) {
this.writeBuffer.poll();
this.callbackBuffer.poll();
}
this.prevBufferLen = 0;
if (this.writeBuffer.size() == 0) {
this.emit(EVENT_DRAIN);
} else {
this.flush();
}
}
private void flush() {
if (this.readyState != ReadyState.CLOSED && this.transport.writable &&
!this.upgrading && this.writeBuffer.size() != 0) {
logger.fine(String.format("flushing %d packets in socket", this.writeBuffer.size()));
this.prevBufferLen = this.writeBuffer.size();
this.transport.send(this.writeBuffer.toArray(new Packet[this.writeBuffer.size()]));
this.emit(EVENT_FLUSH);
}
}
public void write(String msg) {
this.write(msg, null);
}
public void write(String msg, Runnable fn) {
this.send(msg, fn);
}
public void write(byte[] msg) {
this.write(msg, null);
}
public void write(byte[] msg, Runnable fn) {
this.send(msg, fn);
}
/**
* Sends a message.
*
* @param msg
*/
public void send(String msg) {
this.send(msg, null);
}
public void send(byte[] msg) {
this.send(msg, null);
}
/**
* Sends a message.
*
* @param msg
* @param fn callback to be called on drain
*/
public void send(final String msg, final Runnable fn) {
EventThread.exec(new Runnable() {
@Override
public void run() {
Socket.this.sendPacket(Packet.MESSAGE, msg, fn);
}
});
}
public void send(final byte[] msg, final Runnable fn) {
EventThread.exec(new Runnable() {
@Override
public void run() {
Socket.this.sendPacket(Packet.MESSAGE, msg, fn);
}
});
}
private void sendPacket(String type) {
this.sendPacket(new Packet(type), null);
}
private void sendPacket(String type, String data, Runnable fn) {
Packet<String> packet = new Packet<String>(type, data);
sendPacket(packet, fn);
}
private void sendPacket(String type, byte[] data, Runnable fn) {
Packet<byte[]> packet = new Packet<byte[]>(type, data);
sendPacket(packet, fn);
}
private void sendPacket(Packet packet, Runnable fn) {
if (ReadyState.CLOSING == this.readyState || ReadyState.CLOSED == this.readyState) {
return;
}
if (fn == null) {
// ConcurrentLinkedList does not permit `null`.
fn = noop;
}
this.emit(EVENT_PACKET_CREATE, packet);
this.writeBuffer.offer(packet);
this.callbackBuffer.offer(fn);
this.flush();
}
/**
* Disconnects the client.
*
* @return a reference to to this object.
*/
public Socket close() {
EventThread.exec(new Runnable() {
@Override
public void run() {
if (Socket.this.readyState == ReadyState.OPENING || Socket.this.readyState == ReadyState.OPEN) {
Socket.this.readyState = ReadyState.CLOSING;
final Socket self = Socket.this;
final Runnable close = new Runnable() {
@Override
public void run() {
self.onClose("forced close");
logger.fine("socket closing - telling transport to close");
self.transport.close();
}
};
final Listener[] cleanupAndClose = new Listener[1];
cleanupAndClose[0] = new Listener() {
@Override
public void call(Object ...args) {
self.off(EVENT_UPGRADE, cleanupAndClose[0]);
self.off(EVENT_UPGRADE_ERROR, cleanupAndClose[0]);
close.run();
}
};
final Runnable waitForUpgrade = new Runnable() {
@Override
public void run() {
// wait for updade to finish since we can't send packets while pausing a transport
self.once(EVENT_UPGRADE, cleanupAndClose[0]);
self.once(EVENT_UPGRADE_ERROR, cleanupAndClose[0]);
}
};
if (Socket.this.writeBuffer.size() > 0) {
Socket.this.once(EVENT_DRAIN, new Listener() {
@Override
public void call(Object... args) {
if (Socket.this.upgrading) {
waitForUpgrade.run();
} else {
close.run();
}
}
});
} else if (Socket.this.upgrading) {
waitForUpgrade.run();
} else {
close.run();
}
}
}
});
return this;
}
private void onError(Exception err) {
logger.fine(String.format("socket error %s", err));
Socket.priorWebsocketSuccess = false;
this.emit(EVENT_ERROR, err);
this.onClose("transport error", err);
}
private void onClose(String reason) {
this.onClose(reason, null);
}
private void onClose(String reason, Exception desc) {
if (ReadyState.OPENING == this.readyState || ReadyState.OPEN == this.readyState || ReadyState.CLOSING == this.readyState) {
logger.fine(String.format("socket close with reason: %s", reason));
final Socket self = this;
// clear timers
if (this.pingIntervalTimer != null) {
this.pingIntervalTimer.cancel(false);
}
if (this.pingTimeoutTimer != null) {
this.pingTimeoutTimer.cancel(false);
}
if (this.heartbeatScheduler != null) {
this.heartbeatScheduler.shutdown();
}
EventThread.nextTick(new Runnable() {
@Override
public void run() {
self.writeBuffer.clear();
self.callbackBuffer.clear();
self.prevBufferLen = 0;
}
});
// stop event from firing again for transport
this.transport.off(EVENT_CLOSE);
// ensure transport won't stay open
this.transport.close();
// ignore further transport communication
this.transport.off();
// set ready state
this.readyState = ReadyState.CLOSED;
// clear session id
this.id = null;
// emit close events
this.emit(EVENT_CLOSE, reason, desc);
}
}
/*package*/ List<String > filterUpgrades(List<String> upgrades) {
List<String> filteredUpgrades = new ArrayList<String>();
for (String upgrade : upgrades) {
if (this.transports.contains(upgrade)) {
filteredUpgrades.add(upgrade);
}
}
return filteredUpgrades;
}
public String id() {
return this.id;
}
private ScheduledExecutorService getHeartbeatScheduler() {
if (this.heartbeatScheduler == null || this.heartbeatScheduler.isShutdown()) {
this.heartbeatScheduler = Executors.newSingleThreadScheduledExecutor();
}
return this.heartbeatScheduler;
}
public static class Options extends Transport.Options {
/**
* List of transport names.
*/
public String[] transports;
/**
* Whether to upgrade the transport. Defaults to `true`.
*/
public boolean upgrade = true;
public boolean rememberUpgrade;
public String host;
public String query;
private static Options fromURI(URI uri, Options opts) {
if (opts == null) {
opts = new Options();
}
opts.host = uri.getHost();
opts.secure = "https".equals(uri.getScheme()) || "wss".equals(uri.getScheme());
opts.port = uri.getPort();
String query = uri.getRawQuery();
if (query != null) {
opts.query = query;
}
return opts;
}
}
}

153
src/main/java/io/socket/engineio/client/Transport.java Normal file
View File

@@ -0,0 +1,153 @@
package io.socket.engineio.client;
import io.socket.emitter.Emitter;
import io.socket.engineio.parser.Packet;
import io.socket.engineio.parser.Parser;
import io.socket.thread.EventThread;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLContext;
import java.util.Map;
public abstract class Transport extends Emitter {
protected enum ReadyState {
OPENING, OPEN, CLOSED, PAUSED;
@Override
public String toString() {
return super.toString().toLowerCase();
}
}
public static final String EVENT_OPEN = "open";
public static final String EVENT_CLOSE = "close";
public static final String EVENT_PACKET = "packet";
public static final String EVENT_DRAIN = "drain";
public static final String EVENT_ERROR = "error";
public static final String EVENT_REQUEST_HEADERS = "requestHeaders";
public static final String EVENT_RESPONSE_HEADERS = "responseHeaders";
protected static int timestamps = 0;
public boolean writable;
public String name;
public Map<String, String> query;
protected boolean secure;
protected boolean timestampRequests;
protected int port;
protected String path;
protected String hostname;
protected String timestampParam;
protected SSLContext sslContext;
protected Socket socket;
protected HostnameVerifier hostnameVerifier;
protected ReadyState readyState;
public Transport(Options opts) {
this.path = opts.path;
this.hostname = opts.hostname;
this.port = opts.port;
this.secure = opts.secure;
this.query = opts.query;
this.timestampParam = opts.timestampParam;
this.timestampRequests = opts.timestampRequests;
this.sslContext = opts.sslContext;
this.socket = opts.socket;
this.hostnameVerifier = opts.hostnameVerifier;
}
protected Transport onError(String msg, Exception desc) {
// TODO: handle error
Exception err = new EngineIOException(msg, desc);
this.emit(EVENT_ERROR, err);
return this;
}
public Transport open() {
EventThread.exec(new Runnable() {
@Override
public void run() {
if (Transport.this.readyState == ReadyState.CLOSED || Transport.this.readyState == null) {
Transport.this.readyState = ReadyState.OPENING;
Transport.this.doOpen();
}
}
});
return this;
}
public Transport close() {
EventThread.exec(new Runnable() {
@Override
public void run() {
if (Transport.this.readyState == ReadyState.OPENING || Transport.this.readyState == ReadyState.OPEN) {
Transport.this.doClose();
Transport.this.onClose();
}
}
});
return this;
}
public void send(final Packet[] packets) {
EventThread.exec(new Runnable() {
@Override
public void run() {
if (Transport.this.readyState == ReadyState.OPEN) {
Transport.this.write(packets);
} else {
throw new RuntimeException("Transport not open");
}
}
});
}
protected void onOpen() {
this.readyState = ReadyState.OPEN;
this.writable = true;
this.emit(EVENT_OPEN);
}
protected void onData(String data) {
this.onPacket(Parser.decodePacket(data));
}
protected void onData(byte[] data) {
this.onPacket(Parser.decodePacket(data));
}
protected void onPacket(Packet packet) {
this.emit(EVENT_PACKET, packet);
}
protected void onClose() {
this.readyState = ReadyState.CLOSED;
this.emit(EVENT_CLOSE);
}
abstract protected void write(Packet[] packets);
abstract protected void doOpen();
abstract protected void doClose();
public static class Options {
public String hostname;
public String path;
public String timestampParam;
public boolean secure;
public boolean timestampRequests;
public int port = -1;
public int policyPort = -1;
public Map<String, String> query;
public SSLContext sslContext;
public HostnameVerifier hostnameVerifier;
protected Socket socket;
}
}

218
src/main/java/io/socket/engineio/client/transports/Polling.java Normal file
View File

@@ -0,0 +1,218 @@
package io.socket.engineio.client.transports;
import io.socket.engineio.client.Transport;
import io.socket.engineio.parser.Packet;
import io.socket.engineio.parser.Parser;
import io.socket.parseqs.ParseQS;
import io.socket.thread.EventThread;
import io.socket.emitter.Emitter;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;
abstract public class Polling extends Transport {
private static final Logger logger = Logger.getLogger(Polling.class.getName());
public static final String NAME = "polling";
public static final String EVENT_POLL = "poll";
public static final String EVENT_POLL_COMPLETE = "pollComplete";
private boolean polling;
public Polling(Options opts) {
super(opts);
this.name = NAME;
}
protected void doOpen() {
this.poll();
}
public void pause(final Runnable onPause) {
EventThread.exec(new Runnable() {
@Override
public void run() {
final Polling self = Polling.this;
Polling.this.readyState = ReadyState.PAUSED;
final Runnable pause = new Runnable() {
@Override
public void run() {
logger.fine("paused");
self.readyState = ReadyState.PAUSED;
onPause.run();
}
};
if (Polling.this.polling || !Polling.this.writable) {
final int[] total = new int[]{0};
if (Polling.this.polling) {
logger.fine("we are currently polling - waiting to pause");
total[0]++;
Polling.this.once(EVENT_POLL_COMPLETE, new Emitter.Listener() {
@Override
public void call(Object... args) {
logger.fine("pre-pause polling complete");
if (--total[0] == 0) {
pause.run();
}
}
});
}
if (!Polling.this.writable) {
logger.fine("we are currently writing - waiting to pause");
total[0]++;
Polling.this.once(EVENT_DRAIN, new Emitter.Listener() {
@Override
public void call(Object... args) {
logger.fine("pre-pause writing complete");
if (--total[0] == 0) {
pause.run();
}
}
});
}
} else {
pause.run();
}
}
});
}
private void poll() {
logger.fine("polling");
this.polling = true;
this.doPoll();
this.emit(EVENT_POLL);
}
@Override
protected void onData(String data) {
_onData(data);
}
@Override
protected void onData(byte[] data) {
_onData(data);
}
private void _onData(Object data) {
final Polling self = this;
logger.fine(String.format("polling got data %s", data));
Parser.DecodePayloadCallback callback = new Parser.DecodePayloadCallback() {
@Override
public boolean call(Packet packet, int index, int total) {
if (self.readyState == ReadyState.OPENING) {
self.onOpen();
}
if (Packet.CLOSE.equals(packet.type)) {
self.onClose();
return false;
}
self.onPacket(packet);
return true;
}
};
if (data instanceof String) {
@SuppressWarnings("unchecked")
Parser.DecodePayloadCallback<String> _callback = callback;
Parser.decodePayload((String)data, _callback);
} else if (data instanceof byte[]) {
Parser.decodePayload((byte[])data, callback);
}
if (this.readyState != ReadyState.CLOSED) {
this.polling = false;
this.emit(EVENT_POLL_COMPLETE);
if (this.readyState == ReadyState.OPEN) {
this.poll();
} else {
logger.fine(String.format("ignoring poll - transport state '%s'", this.readyState));
}
}
}
protected void doClose() {
final Polling self = this;
Emitter.Listener close = new Emitter.Listener() {
@Override
public void call(Object... args) {
logger.fine("writing close packet");
self.write(new Packet[] {new Packet(Packet.CLOSE)});
}
};
if (this.readyState == ReadyState.OPEN) {
logger.fine("transport open - closing");
close.call();
} else {
// in case we're trying to close while
// handshaking is in progress (engine.io-client GH-164)
logger.fine("transport not open - deferring close");
this.once(EVENT_OPEN, close);
}
}
protected void write(Packet[] packets) {
final Polling self = this;
this.writable = false;
final Runnable callbackfn = new Runnable() {
@Override
public void run() {
self.writable = true;
self.emit(EVENT_DRAIN);
}
};
Parser.encodePayload(packets, new Parser.EncodeCallback<byte[]>() {
@Override
public void call(byte[] data) {
self.doWrite(data, callbackfn);
}
});
}
protected String uri() {
Map<String, String> query = this.query;
if (query == null) {
query = new HashMap<String, String>();
}
String schema = this.secure ? "https" : "http";
String port = "";
if (this.timestampRequests) {
query.put(this.timestampParam, String.valueOf(new Date().getTime()) + "-" + Transport.timestamps++);
}
String _query = ParseQS.encode(query);
if (this.port > 0 && (("https".equals(schema) && this.port != 443)
|| ("http".equals(schema) && this.port != 80))) {
port = ":" + this.port;
}
if (_query.length() > 0) {
_query = "?" + _query;
}
return schema + "://" + this.hostname + port + this.path + _query;
}
abstract protected void doWrite(byte[] data, Runnable fn);
abstract protected void doPoll();
}

323
src/main/java/io/socket/engineio/client/transports/PollingXHR.java Normal file
View File

@@ -0,0 +1,323 @@
package io.socket.engineio.client.transports;
import io.socket.emitter.Emitter;
import io.socket.thread.EventThread;
import io.socket.engineio.client.Transport;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.logging.Logger;
public class PollingXHR extends Polling {
private static final Logger logger = Logger.getLogger(PollingXHR.class.getName());
private Request sendXhr;
private Request pollXhr;
public PollingXHR(Transport.Options opts) {
super(opts);
}
protected Request request() {
return this.request(null);
}
protected Request request(Request.Options opts) {
if (opts == null) {
opts = new Request.Options();
}
opts.uri = this.uri();
opts.sslContext = this.sslContext;
opts.hostnameVerifier = this.hostnameVerifier;
Request req = new Request(opts);
final PollingXHR self = this;
req.on(Request.EVENT_REQUEST_HEADERS, new Emitter.Listener() {
@Override
public void call(Object... args) {
// Never execute asynchronously for support to modify headers.
self.emit(Transport.EVENT_REQUEST_HEADERS, args[0]);
}
}).on(Request.EVENT_RESPONSE_HEADERS, new Emitter.Listener() {
@Override
public void call(final Object... args) {
EventThread.exec(new Runnable() {
@Override
public void run() {
self.emit(Transport.EVENT_RESPONSE_HEADERS, args[0]);
}
});
}
});
return req;
}
@Override
protected void doWrite(byte[] data, final Runnable fn) {
Request.Options opts = new Request.Options();
opts.method = "POST";
opts.data = data;
Request req = this.request(opts);
final PollingXHR self = this;
req.on(Request.EVENT_SUCCESS, new Emitter.Listener() {
@Override
public void call(Object... args) {
EventThread.exec(new Runnable() {
@Override
public void run() {
fn.run();
}
});
}
});
req.on(Request.EVENT_ERROR, new Emitter.Listener() {
@Override
public void call(final Object... args) {
EventThread.exec(new Runnable() {
@Override
public void run() {
Exception err = args.length > 0 && args[0] instanceof Exception ? (Exception)args[0] : null;
self.onError("xhr post error", err);
}
});
}
});
req.create();
this.sendXhr = req;
}
@Override
protected void doPoll() {
logger.fine("xhr poll");
Request req = this.request();
final PollingXHR self = this;
req.on(Request.EVENT_DATA, new Emitter.Listener() {
@Override
public void call(final Object... args) {
EventThread.exec(new Runnable() {
@Override
public void run() {
Object arg = args.length > 0 ? args[0] : null;
if (arg instanceof String) {
self.onData((String)arg);
} else if (arg instanceof byte[]) {
self.onData((byte[])arg);
}
}
});
}
});
req.on(Request.EVENT_ERROR, new Emitter.Listener() {
@Override
public void call(final Object... args) {
EventThread.exec(new Runnable() {
@Override
public void run() {
Exception err = args.length > 0 && args[0] instanceof Exception ? (Exception) args[0] : null;
self.onError("xhr poll error", err);
}
});
}
});
req.create();
this.pollXhr = req;
}
public static class Request extends Emitter {
public static final String EVENT_SUCCESS = "success";
public static final String EVENT_DATA = "data";
public static final String EVENT_ERROR = "error";
public static final String EVENT_REQUEST_HEADERS = "requestHeaders";
public static final String EVENT_RESPONSE_HEADERS = "responseHeaders";
private String method;
private String uri;
// data is always a binary
private byte[] data;
private SSLContext sslContext;
private HttpURLConnection xhr;
private HostnameVerifier hostnameVerifier;
public Request(Options opts) {
this.method = opts.method != null ? opts.method : "GET";
this.uri = opts.uri;
this.data = opts.data;
this.sslContext = opts.sslContext;
this.hostnameVerifier = opts.hostnameVerifier;
}
public void create() {
final Request self = this;
try {
logger.fine(String.format("xhr open %s: %s", this.method, this.uri));
URL url = new URL(this.uri);
xhr = (HttpURLConnection)url.openConnection();
xhr.setRequestMethod(this.method);
} catch (IOException e) {
this.onError(e);
return;
}
xhr.setConnectTimeout(10000);
if (xhr instanceof HttpsURLConnection) {
if (this.sslContext != null) {
((HttpsURLConnection)xhr).setSSLSocketFactory(this.sslContext.getSocketFactory());
}
if (this.hostnameVerifier != null) {
((HttpsURLConnection)xhr).setHostnameVerifier(this.hostnameVerifier);
}
}
Map<String, List<String>> headers = new TreeMap<String, List<String>>(String.CASE_INSENSITIVE_ORDER);
if ("POST".equals(this.method)) {
xhr.setDoOutput(true);
headers.put("Content-type", new LinkedList<String>(Arrays.asList("application/octet-stream")));
}
self.onRequestHeaders(headers);
for (Map.Entry<String, List<String>> header : headers.entrySet()) {
for (String v : header.getValue()){
xhr.addRequestProperty(header.getKey(), v);
}
}
logger.fine(String.format("sending xhr with url %s | data %s", this.uri, this.data));
new Thread(new Runnable() {
@Override
public void run() {
OutputStream output = null;
try {
if (self.data != null) {
xhr.setFixedLengthStreamingMode(self.data.length);
output = new BufferedOutputStream(xhr.getOutputStream());
output.write(self.data);
output.flush();
}
Map<String, List<String>> headers = xhr.getHeaderFields();
self.onResponseHeaders(headers);
final int statusCode = xhr.getResponseCode();
if (HttpURLConnection.HTTP_OK == statusCode) {
self.onLoad();
} else {
self.onError(new IOException(Integer.toString(statusCode)));
}
} catch (IOException e) {
self.onError(e);
} finally {
try {
if (output != null) output.close();
} catch (IOException e) {}
}
}
}).start();
}
private void onSuccess() {
this.emit(EVENT_SUCCESS);
this.cleanup();
}
private void onData(String data) {
this.emit(EVENT_DATA, data);
this.onSuccess();
}
private void onData(byte[] data) {
this.emit(EVENT_DATA, data);
this.onSuccess();
}
private void onError(Exception err) {
this.emit(EVENT_ERROR, err);
this.cleanup();
}
private void onRequestHeaders(Map<String, List<String>> headers) {
this.emit(EVENT_REQUEST_HEADERS, headers);
}
private void onResponseHeaders(Map<String, List<String>> headers) {
this.emit(EVENT_RESPONSE_HEADERS, headers);
}
private void cleanup() {
if (xhr == null) {
return;
}
xhr.disconnect();
xhr = null;
}
private void onLoad() {
InputStream input = null;
BufferedReader reader = null;
String contentType = xhr.getContentType();
try {
if ("application/octet-stream".equalsIgnoreCase(contentType)) {
input = new BufferedInputStream(this.xhr.getInputStream());
List<byte[]> buffers = new ArrayList<byte[]>();
int capacity = 0;
int len = 0;
byte[] buffer = new byte[1024];
while ((len = input.read(buffer)) > 0) {
byte[] _buffer = new byte[len];
System.arraycopy(buffer, 0, _buffer, 0, len);
buffers.add(_buffer);
capacity += len;
}
ByteBuffer data = ByteBuffer.allocate(capacity);
for (byte[] b : buffers) {
data.put(b);
}
this.onData(data.array());
} else {
String line;
StringBuilder data = new StringBuilder();
reader = new BufferedReader(new InputStreamReader(xhr.getInputStream()));
while ((line = reader.readLine()) != null) {
data.append(line);
}
this.onData(data.toString());
}
} catch (IOException e) {
this.onError(e);
} finally {
try {
if (input != null) input.close();
} catch (IOException e) {}
try {
if (reader != null) reader.close();
} catch (IOException e) {}
}
}
public void abort() {
this.cleanup();
}
public static class Options {
public String uri;
public String method;
public byte[] data;
public SSLContext sslContext;
public HostnameVerifier hostnameVerifier;
}
}
}

215
src/main/java/io/socket/engineio/client/transports/WebSocket.java Normal file
View File

@@ -0,0 +1,215 @@
package io.socket.engineio.client.transports;
import io.socket.engineio.client.Transport;
import io.socket.engineio.parser.Packet;
import io.socket.engineio.parser.Parser;
import io.socket.parseqs.ParseQS;
import io.socket.thread.EventThread;
import com.squareup.okhttp.OkHttpClient;
import com.squareup.okhttp.Request;
import com.squareup.okhttp.Response;
import com.squareup.okhttp.ws.WebSocket.PayloadType;
import com.squareup.okhttp.ws.WebSocketCall;
import com.squareup.okhttp.ws.WebSocketListener;
import okio.Buffer;
import okio.BufferedSource;
import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.util.*;
import java.util.logging.Logger;
import static com.squareup.okhttp.ws.WebSocket.PayloadType.BINARY;
import static com.squareup.okhttp.ws.WebSocket.PayloadType.TEXT;
public class WebSocket extends Transport {
public static final String NAME = "websocket";
private static final Logger logger = Logger.getLogger(PollingXHR.class.getName());
private com.squareup.okhttp.ws.WebSocket ws;
private WebSocketCall wsCall;
public WebSocket(Options opts) {
super(opts);
this.name = NAME;
}
protected void doOpen() {
Map<String, List<String>> headers = new TreeMap<String, List<String>>(String.CASE_INSENSITIVE_ORDER);
this.emit(EVENT_REQUEST_HEADERS, headers);
final WebSocket self = this;
final OkHttpClient client = new OkHttpClient();
if (this.sslContext != null) {
SSLSocketFactory factory = sslContext.getSocketFactory();// (SSLSocketFactory) SSLSocketFactory.getDefault();
client.setSslSocketFactory(factory);
}
if (this.hostnameVerifier != null) {
client.setHostnameVerifier(this.hostnameVerifier);
}
Request.Builder builder = new Request.Builder().url(uri());
for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
for (String v : entry.getValue()) {
builder.addHeader(entry.getKey(), v);
}
}
final Request request = builder.build();
wsCall = WebSocketCall.create(client, request);
wsCall.enqueue(new WebSocketListener() {
@Override
public void onOpen(com.squareup.okhttp.ws.WebSocket webSocket, Response response) {
ws = webSocket;
final Map<String, List<String>> headers = response.headers().toMultimap();
EventThread.exec(new Runnable() {
@Override
public void run() {
self.emit(EVENT_RESPONSE_HEADERS, headers);
self.onOpen();
}
});
}
@Override
public void onMessage(BufferedSource payload, final PayloadType type) throws IOException {
Object data = null;
switch (type) {
case TEXT:
data = payload.readUtf8();
break;
case BINARY:
data = payload.readByteArray();
break;
default:
EventThread.exec(new Runnable() {
@Override
public void run() {
self.onError("Unknown payload type: " + type, new IllegalStateException());
}
});
}
payload.close();
final Object finalData = data;
EventThread.exec(new Runnable() {
@Override
public void run() {
if (finalData == null) {
return;
}
if (finalData instanceof String) {
self.onData((String) finalData);
} else {
self.onData((byte[]) finalData);
}
}
});
}
@Override
public void onPong(Buffer payload) {
}
@Override
public void onClose(int code, String reason) {
EventThread.exec(new Runnable() {
@Override
public void run() {
self.onClose();
}
});
}
@Override
public void onFailure(final IOException e, final Response response) {
EventThread.exec(new Runnable() {
@Override
public void run() {
self.onError("websocket error", e);
}
});
}
});
client.getDispatcher().getExecutorService().shutdown();
}
protected void write(Packet[] packets) {
final WebSocket self = this;
this.writable = false;
for (Packet packet : packets) {
Parser.encodePacket(packet, new Parser.EncodeCallback() {
@Override
public void call(Object packet) {
try {
if (packet instanceof String) {
self.ws.sendMessage(TEXT, new Buffer().writeUtf8((String) packet));
} else if (packet instanceof byte[]) {
self.ws.sendMessage(BINARY, new Buffer().write((byte[]) packet));
}
} catch (IOException e) {
logger.fine("websocket closed before onclose event");
}
}
});
}
final Runnable ondrain = new Runnable() {
@Override
public void run() {
self.writable = true;
self.emit(EVENT_DRAIN);
}
};
// fake drain
// defer to next tick to allow Socket to clear writeBuffer
EventThread.nextTick(ondrain);
}
@Override
protected void onClose() {
super.onClose();
}
protected void doClose() {
if (wsCall != null) {
wsCall.cancel();
}
if (ws != null) {
try {
ws.close(1000, "");
} catch (IOException e) {
// websocket already closed
} catch (IllegalStateException e) {
// websocket already closed
}
}
}
protected String uri() {
Map<String, String> query = this.query;
if (query == null) {
query = new HashMap<String, String>();
}
String schema = this.secure ? "wss" : "ws";
String port = "";
if (this.port > 0 && (("wss".equals(schema) && this.port != 443)
|| ("ws".equals(schema) && this.port != 80))) {
port = ":" + this.port;
}
if (this.timestampRequests) {
query.put(this.timestampParam, String.valueOf(new Date().getTime()));
}
String _query = ParseQS.encode(query);
if (_query.length() > 0) {
_query = "?" + _query;
}
return schema + "://" + this.hostname + port + this.path + _query;
}
}

27
src/main/java/io/socket/engineio/parser/Packet.java Normal file
View File

@@ -0,0 +1,27 @@
package io.socket.engineio.parser;
public class Packet<T> {
static final public String OPEN = "open";
static final public String CLOSE = "close";
static final public String PING = "ping";
static final public String PONG = "pong";
static final public String UPGRADE = "upgrade";
static final public String MESSAGE = "message";
static final public String NOOP = "noop";
static final public String ERROR = "error";
public String type;
public T data;
public Packet(String type) {
this(type, null);
}
public Packet(String type, T data) {
this.type = type;
this.data = data;
}
}

316
src/main/java/io/socket/engineio/parser/Parser.java Normal file
View File

@@ -0,0 +1,316 @@
package io.socket.engineio.parser;
import io.socket.utf8.UTF8;
import io.socket.utf8.UTF8Exception;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class Parser {
private static final int MAX_INT_CHAR_LENGTH = String.valueOf(Integer.MAX_VALUE).length();
public static final int protocol = 3;
private static final Map<String, Integer> packets = new HashMap<String, Integer>() {{
put(Packet.OPEN, 0);
put(Packet.CLOSE, 1);
put(Packet.PING, 2);
put(Packet.PONG, 3);
put(Packet.MESSAGE, 4);
put(Packet.UPGRADE, 5);
put(Packet.NOOP, 6);
}};
private static final Map<Integer, String> packetslist = new HashMap<Integer, String>();
static {
for (Map.Entry<String, Integer> entry : packets.entrySet()) {
packetslist.put(entry.getValue(), entry.getKey());
}
}
private static Packet<String> err = new Packet<String>(Packet.ERROR, "parser error");
private Parser() {}
public static void encodePacket(Packet packet, EncodeCallback callback) {
encodePacket(packet, false, callback);
}
public static void encodePacket(Packet packet, boolean utf8encode, EncodeCallback callback) {
if (packet.data instanceof byte[]) {
@SuppressWarnings("unchecked")
Packet<byte[]> _packet = packet;
@SuppressWarnings("unchecked")
EncodeCallback<byte[]> _callback = callback;
encodeByteArray(_packet, _callback);
return;
}
String encoded = String.valueOf(packets.get(packet.type));
if (null != packet.data) {
encoded += utf8encode ? UTF8.encode(String.valueOf(packet.data)) : String.valueOf(packet.data);
}
@SuppressWarnings("unchecked")
EncodeCallback<String> _callback = callback;
_callback.call(encoded);
}
private static void encodeByteArray(Packet<byte[]> packet, EncodeCallback<byte[]> callback) {
byte[] data = packet.data;
byte[] resultArray = new byte[1 + data.length];
resultArray[0] = packets.get(packet.type).byteValue();
System.arraycopy(data, 0, resultArray, 1, data.length);
callback.call(resultArray);
}
public static Packet<String> decodePacket(String data) {
return decodePacket(data, false);
}
public static Packet<String> decodePacket(String data, boolean utf8decode) {
int type;
try {
type = Character.getNumericValue(data.charAt(0));
} catch (IndexOutOfBoundsException e) {
type = -1;
}
if (utf8decode) {
try {
data = UTF8.decode(data);
} catch (UTF8Exception e) {
return err;
}
}
if (type < 0 || type >= packetslist.size()) {
return err;
}
if (data.length() > 1) {
return new Packet<String>(packetslist.get(type), data.substring(1));
} else {
return new Packet<String>(packetslist.get(type));
}
}
public static Packet<byte[]> decodePacket(byte[] data) {
int type = data[0];
byte[] intArray = new byte[data.length - 1];
System.arraycopy(data, 1, intArray, 0, intArray.length);
return new Packet<byte[]>(packetslist.get(type), intArray);
}
public static void encodePayload(Packet[] packets, EncodeCallback<byte[]> callback) {
if (packets.length == 0) {
callback.call(new byte[0]);
return;
}
final ArrayList<byte[]> results = new ArrayList<byte[]>(packets.length);
for (Packet packet : packets) {
encodePacket(packet, true, new EncodeCallback() {
@Override
public void call(Object packet) {
if (packet instanceof String) {
String encodingLength = String.valueOf(((String) packet).length());
byte[] sizeBuffer = new byte[encodingLength.length() + 2];
sizeBuffer[0] = (byte)0; // is a string
for (int i = 0; i < encodingLength.length(); i ++) {
sizeBuffer[i + 1] = (byte)Character.getNumericValue(encodingLength.charAt(i));
}
sizeBuffer[sizeBuffer.length - 1] = (byte)255;
results.add(Buffer.concat(new byte[][] {sizeBuffer, stringToByteArray((String)packet)}));
return;
}
String encodingLength = String.valueOf(((byte[])packet).length);
byte[] sizeBuffer = new byte[encodingLength.length() + 2];
sizeBuffer[0] = (byte)1; // is binary
for (int i = 0; i < encodingLength.length(); i ++) {
sizeBuffer[i + 1] = (byte)Character.getNumericValue(encodingLength.charAt(i));
}
sizeBuffer[sizeBuffer.length - 1] = (byte)255;
results.add(Buffer.concat(new byte[][] {sizeBuffer, (byte[])packet}));
}
});
}
callback.call(Buffer.concat(results.toArray(new byte[results.size()][])));
}
public static void decodePayload(String data, DecodePayloadCallback<String> callback) {
if (data == null || data.length() == 0) {
callback.call(err, 0, 1);
return;
}
StringBuilder length = new StringBuilder();
for (int i = 0, l = data.length(); i < l; i++) {
char chr = data.charAt(i);
if (':' != chr) {
length.append(chr);
} else {
int n;
try {
n = Integer.parseInt(length.toString());
} catch (NumberFormatException e) {
callback.call(err, 0, 1);
return;
}
String msg;
try {
msg = data.substring(i + 1, i + 1 + n);
} catch (IndexOutOfBoundsException e) {
callback.call(err, 0, 1);
return;
}
if (msg.length() != 0) {
Packet<String> packet = decodePacket(msg, true);
if (err.type.equals(packet.type) && err.data.equals(packet.data)) {
callback.call(err, 0, 1);
return;
}
boolean ret = callback.call(packet, i + n, l);
if (!ret) return;
}
i += n;
length = new StringBuilder();
}
}
if (length.length() > 0) {
callback.call(err, 0, 1);
}
}
public static void decodePayload(byte[] data, DecodePayloadCallback callback) {
ByteBuffer bufferTail = ByteBuffer.wrap(data);
List<Object> buffers = new ArrayList<Object>();
while (bufferTail.capacity() > 0) {
StringBuilder strLen = new StringBuilder();
boolean isString = (bufferTail.get(0) & 0xFF) == 0;
boolean numberTooLong = false;
for (int i = 1; ; i++) {
int b = bufferTail.get(i) & 0xFF;
if (b == 255) break;
// supports only integer
if (strLen.length() > MAX_INT_CHAR_LENGTH) {
numberTooLong = true;
break;
}
strLen.append(b);
}
if (numberTooLong) {
@SuppressWarnings("unchecked")
DecodePayloadCallback<String> _callback = callback;
_callback.call(err, 0, 1);
return;
}
bufferTail.position(strLen.length() + 1);
bufferTail = bufferTail.slice();
int msgLength = Integer.parseInt(strLen.toString());
bufferTail.position(1);
bufferTail.limit(msgLength + 1);
byte[] msg = new byte[bufferTail.remaining()];
bufferTail.get(msg);
if (isString) {
buffers.add(byteArrayToString(msg));
} else {
buffers.add(msg);
}
bufferTail.clear();
bufferTail.position(msgLength + 1);
bufferTail = bufferTail.slice();
}
int total = buffers.size();
for (int i = 0; i < total; i++) {
Object buffer = buffers.get(i);
if (buffer instanceof String) {
@SuppressWarnings("unchecked")
DecodePayloadCallback<String> _callback = callback;
_callback.call(decodePacket((String)buffer, true), i, total);
} else if (buffer instanceof byte[]) {
@SuppressWarnings("unchecked")
DecodePayloadCallback<byte[]> _callback = callback;
_callback.call(decodePacket((byte[])buffer), i, total);
}
}
}
private static String byteArrayToString(byte[] bytes) {
StringBuilder builder = new StringBuilder();
for (byte b : bytes) {
builder.appendCodePoint(b & 0xFF);
}
return builder.toString();
}
private static byte[] stringToByteArray(String string) {
int len = string.length();
byte[] bytes = new byte[len];
for (int i = 0; i < len; i++) {
bytes[i] = (byte)Character.codePointAt(string, i);
}
return bytes;
}
public static interface EncodeCallback<T> {
public void call(T data);
}
public static interface DecodePayloadCallback<T> {
public boolean call(Packet<T> packet, int index, int total);
}
}
class Buffer {
private Buffer() {}
public static byte[] concat(byte[][] list) {
int length = 0;
for (byte[] buf : list) {
length += buf.length;
}
return concat(list, length);
}
public static byte[] concat(byte[][] list, int length) {
if (list.length == 0) {
return new byte[0];
} else if (list.length == 1) {
return list[0];
}
ByteBuffer buffer = ByteBuffer.allocate(length);
for (byte[] buf : list) {
buffer.put(buf);
}
return buffer.array();
}
}

33
src/main/java/io/socket/global/Global.java Normal file
View File

@@ -0,0 +1,33 @@
package io.socket.global;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.net.URLEncoder;
public class Global {
private Global() {}
public static String encodeURIComponent(String str) {
try {
return URLEncoder.encode(str, "UTF-8")
.replace("+", "%20")
.replace("%21", "!")
.replace("%27", "'")
.replace("%28", "(")
.replace("%29", ")")
.replace("%7E", "~");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e);
}
}
public static String decodeURIComponent(String str) {
try {
return URLDecoder.decode(str, "UTF-8");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException(e);
}
}
}

33
src/main/java/io/socket/parseqs/ParseQS.java Normal file
View File

@@ -0,0 +1,33 @@
package io.socket.parseqs;
import io.socket.global.Global;
import java.util.HashMap;
import java.util.Map;
public class ParseQS {
private ParseQS() {}
public static String encode(Map<String, String> obj) {
StringBuilder str = new StringBuilder();
for (Map.Entry<String, String> entry : obj.entrySet()) {
if (str.length() > 0) str.append("&");
str.append(Global.encodeURIComponent(entry.getKey())).append("=")
.append(Global.encodeURIComponent(entry.getValue()));
}
return str.toString();
}
public static Map<String, String> decode(String qs) {
Map<String, String> qry = new HashMap<String, String>();
String[] pairs = qs.split("&");
for (String _pair : pairs) {
String[] pair = _pair.split("=");
qry.put(Global.decodeURIComponent(pair[0]),
pair.length > 1 ? Global.decodeURIComponent(pair[1]) : "");
}
return qry;
}
}

89
src/main/java/io/socket/thread/EventThread.java Normal file
View File

@@ -0,0 +1,89 @@
package io.socket.thread;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
/**
* The thread for event loop. All non-background tasks run within this thread.
*/
public class EventThread extends Thread {
private static final ThreadFactory THREAD_FACTORY = new ThreadFactory() {
@Override
public Thread newThread(Runnable runnable) {
thread = new EventThread(runnable);
thread.setName("EventThread");
return thread;
}
};
private static EventThread thread;
private static ExecutorService service;
private static int counter = 0;
private EventThread(Runnable runnable) {
super(runnable);
}
/**
* check if the current thread is EventThread.
*
* @return true if the current thread is EventThread.
*/
public static boolean isCurrent() {
return currentThread() == thread;
}
/**
* Executes a task in EventThread.
*
* @param task
*/
public static void exec(Runnable task) {
if (isCurrent()) {
task.run();
} else {
nextTick(task);
}
}
/**
* Executes a task on the next loop in EventThread.
*
* @param task
*/
public static void nextTick(final Runnable task) {
ExecutorService executor;
synchronized (EventThread.class) {
counter++;
if (service == null) {
service = Executors.newSingleThreadExecutor(THREAD_FACTORY);
}
executor = service;
}
executor.execute(new Runnable() {
@Override
public void run() {
try {
task.run();
} finally {
synchronized (EventThread.class) {
counter--;
if (counter == 0) {
service.shutdown();
service = null;
thread = null;
}
}
}
}
});
}
}

164
src/main/java/io/socket/utf8/UTF8.java Normal file
View File

@@ -0,0 +1,164 @@
package io.socket.utf8;
import java.util.ArrayList;
import java.util.List;
/**
* UTF-8 encoder/decoder ported from utf8.js.
*
* @see <a href="https://github.com/mathiasbynens/utf8.js">https://github.com/mathiasbynens/utf8.js</a>
*/
public class UTF8 {
private static int[] byteArray;
private static int byteCount;
private static int byteIndex;
public static String encode(String string) {
int[] codePoints = uc2decode(string);
int length = codePoints.length;
int index = -1;
int codePoint;
StringBuilder byteString = new StringBuilder();
while (++index < length) {
codePoint = codePoints[index];
byteString.append(encodeCodePoint(codePoint));
}
return byteString.toString();
}
public static String decode(String byteString) throws UTF8Exception {
byteArray = uc2decode(byteString);
byteCount = byteArray.length;
byteIndex = 0;
List<Integer> codePoints = new ArrayList<Integer>();
int tmp;
while ((tmp = decodeSymbol()) != -1) {
codePoints.add(tmp);
}
return ucs2encode(listToArray(codePoints));
}
private static int[] uc2decode(String string) {
int length = string.length();
int[] output = new int[string.codePointCount(0, length)];
int counter = 0;
int value;
for (int i = 0; i < length; i += Character.charCount(value)) {
value = string.codePointAt(i);
output[counter++] = value;
}
return output;
}
private static String encodeCodePoint(int codePoint) {
StringBuilder symbol = new StringBuilder();
if ((codePoint & 0xFFFFFF80) == 0) {
return symbol.append(Character.toChars(codePoint)).toString();
}
if ((codePoint & 0xFFFFF800) == 0) {
symbol.append(Character.toChars(((codePoint >> 6) & 0x1F) | 0xC0));
} else if ((codePoint & 0xFFFF0000) == 0) {
symbol.append(Character.toChars(((codePoint >> 12) & 0x0F) | 0xE0));
symbol.append(createByte(codePoint, 6));
} else if ((codePoint & 0xFFE00000) == 0) {
symbol.append(Character.toChars(((codePoint >> 18) & 0x07) | 0xF0));
symbol.append(createByte(codePoint, 12));
symbol.append(createByte(codePoint, 6));
}
symbol.append(Character.toChars((codePoint & 0x3F) | 0x80));
return symbol.toString();
}
private static char[] createByte(int codePoint, int shift) {
return Character.toChars(((codePoint >> shift) & 0x3F) | 0x80);
}
private static int decodeSymbol() throws UTF8Exception {
int byte1;
int byte2;
int byte3;
int byte4;
int codePoint;
if (byteIndex > byteCount) {
throw new UTF8Exception("Invalid byte index");
}
if (byteIndex == byteCount) {
return -1;
}
byte1 = byteArray[byteIndex] & 0xFF;
byteIndex++;
if ((byte1 & 0x80) == 0) {
return byte1;
}
if ((byte1 & 0xE0) == 0xC0) {
byte2 = readContinuationByte();
codePoint = ((byte1 & 0x1F) << 6) | byte2;
if (codePoint >= 0x80) {
return codePoint;
} else {
throw new UTF8Exception("Invalid continuation byte");
}
}
if ((byte1 & 0xF0) == 0xE0) {
byte2 = readContinuationByte();
byte3 = readContinuationByte();
codePoint = ((byte1 & 0x0F) << 12) | (byte2 << 6) | byte3;
if (codePoint >= 0x0800) {
return codePoint;
} else {
throw new UTF8Exception("Invalid continuation byte");
}
}
if ((byte1 & 0xF8) == 0xF0) {
byte2 = readContinuationByte();
byte3 = readContinuationByte();
byte4 = readContinuationByte();
codePoint = ((byte1 & 0x0F) << 0x12) | (byte2 << 0x0C) | (byte3 << 0x06) | byte4;
if (codePoint >= 0x010000 && codePoint <= 0x10FFFF) {
return codePoint;
}
}
throw new UTF8Exception("Invalid continuation byte");
}
private static int readContinuationByte() throws UTF8Exception {
if (byteIndex >= byteCount) {
throw new UTF8Exception("Invalid byte index");
}
int continuationByte = byteArray[byteIndex] & 0xFF;
byteIndex++;
if ((continuationByte & 0xC0) == 0x80) {
return continuationByte & 0x3F;
}
throw new UTF8Exception("Invalid continuation byte");
}
private static String ucs2encode(int[] array) {
StringBuilder output = new StringBuilder();
for (int value : array) {
output.appendCodePoint(value);
}
return output.toString();
}
private static int[] listToArray(List<Integer> list) {
int size = list.size();
int[] array = new int[size];
for (int i = 0; i < size; i++) {
array[i] = list.get(i);
}
return array;
}
}

24
src/main/java/io/socket/utf8/UTF8Exception.java Normal file
View File

@@ -0,0 +1,24 @@
package io.socket.utf8;
import java.io.IOException;
public class UTF8Exception extends IOException {
public String data;
public UTF8Exception() {
super();
}
public UTF8Exception(String message) {
super(message);
}
public UTF8Exception(String message, Throwable cause) {
super(message, cause);
}
public UTF8Exception(Throwable cause) {
super(cause);
}
}