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

thread constraint

Browse Source
This commit is contained in:
Naoyuki Kanezawa
2013-05-07 01:47:59 +09:00
parent ee4f0622e2
commit 8257f68bab
3 changed files with 192 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/com/github/nkzawa/socketio/client/IO.java
View File

@@ -7,15 +7,15 @@ import java.net.MalformedURLException;
import java.net.URI; import java.net.URI;
import java.net.URISyntaxException; import java.net.URISyntaxException;
import java.net.URL; import java.net.URL;
import java.util.HashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.Map;
public class IO { public class IO {
private static final Map<String, Manager> managers = new HashMap<String, Manager>(); private static final ConcurrentHashMap<String, Manager> managers = new ConcurrentHashMap<String, Manager>();
public static int protocol = Parser.protocol; public static int protocol = Parser.protocol;
private IO() {} private IO() {}
public static Socket socket(String uri) throws URISyntaxException { public static Socket socket(String uri) throws URISyntaxException {
@@ -49,7 +49,7 @@ public class IO {
} else { } else {
String id = Url.extractId(parsed); String id = Url.extractId(parsed);
if (!managers.containsKey(id)) { if (!managers.containsKey(id)) {
managers.put(id, new Manager(href, opts)); managers.putIfAbsent(id, new Manager(href, opts));
} }
io = managers.get(id); io = managers.get(id);
} }

73
src/main/java/com/github/nkzawa/socketio/client/Manager.java
View File

@@ -1,14 +1,14 @@
package com.github.nkzawa.socketio.client; package com.github.nkzawa.socketio.client;
import com.github.nkzawa.emitter.Emitter; import com.github.nkzawa.emitter.Emitter;
import com.github.nkzawa.engineio.client.EventThread;
import com.github.nkzawa.socketio.parser.Packet; import com.github.nkzawa.socketio.parser.Packet;
import com.github.nkzawa.socketio.parser.Parser; import com.github.nkzawa.socketio.parser.Parser;
import java.net.URI; import java.net.URI;
import java.util.Map; import java.util.LinkedList;
import java.util.Queue; import java.util.Queue;
import java.util.concurrent.*; import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger; import java.util.logging.Logger;
public class Manager extends Emitter { public class Manager extends Emitter {
@@ -38,15 +38,20 @@ public class Manager extends Emitter {
private long _reconnectionDelay; private long _reconnectionDelay;
private long _reconnectionDelayMax; private long _reconnectionDelayMax;
private long _timeout; private long _timeout;
private AtomicInteger connected = new AtomicInteger(); private int connected;
private AtomicInteger attempts = new AtomicInteger(); private int attempts;
private Map<String, Socket> nsps = new ConcurrentHashMap<String, Socket>(); private Queue<On.Handle> subs = new LinkedList<On.Handle>();
private Queue<On.Handle> subs = new ConcurrentLinkedQueue<On.Handle>();
private com.github.nkzawa.engineio.client.Socket engine; private com.github.nkzawa.engineio.client.Socket engine;
/**
* This HashMap can be accessed from outside of EventThread.
*/
private ConcurrentHashMap<String, Socket> nsps = new ConcurrentHashMap<String, Socket>();
private ScheduledExecutorService timeoutScheduler = Executors.newSingleThreadScheduledExecutor(); private ScheduledExecutorService timeoutScheduler = Executors.newSingleThreadScheduledExecutor();
private ScheduledExecutorService reconnectScheduler = Executors.newSingleThreadScheduledExecutor(); private ScheduledExecutorService reconnectScheduler = Executors.newSingleThreadScheduledExecutor();
public Manager(URI uri, IO.Options opts) { public Manager(URI uri, IO.Options opts) {
opts = initOptions(opts); opts = initOptions(opts);
this.engine = new Engine(uri, opts); this.engine = new Engine(uri, opts);
@@ -122,12 +127,15 @@ public class Manager extends Emitter {
} }
public Manager open(final OpenCallback fn) { public Manager open(final OpenCallback fn) {
if (this.readyState == OPEN && !this.reconnecting) return this; EventThread.exec(new Runnable() {
@Override
public void run() {
if (Manager.this.readyState == OPEN && !Manager.this.reconnecting) return;
final com.github.nkzawa.engineio.client.Socket socket = this.engine; final com.github.nkzawa.engineio.client.Socket socket = Manager.this.engine;
final Manager self = this; final Manager self = Manager.this;
this.readyState = OPENING; Manager.this.readyState = OPENING;
final On.Handle openSub = On.on(socket, Engine.EVENT_OPEN, new Listener() { final On.Handle openSub = On.on(socket, Engine.EVENT_OPEN, new Listener() {
@Override @Override
@@ -151,11 +159,14 @@ public class Manager extends Emitter {
} }
}); });
if (this._timeout >= 0) { if (Manager.this._timeout >= 0) {
final long timeout = this._timeout; final long timeout = Manager.this._timeout;
logger.fine(String.format("connection attempt will timeout after %d", timeout)); logger.fine(String.format("connection attempt will timeout after %d", timeout));
final Future timer = timeoutScheduler.schedule(new Runnable() { final Future timer = timeoutScheduler.schedule(new Runnable() {
@Override
public void run() {
EventThread.exec(new Runnable() {
@Override @Override
public void run() { public void run() {
logger.fine(String.format("connect attempt timed out after %d", timeout)); logger.fine(String.format("connect attempt timed out after %d", timeout));
@@ -164,6 +175,8 @@ public class Manager extends Emitter {
socket.emit(Engine.EVENT_ERROR, new SocketIOException("timeout")); socket.emit(Engine.EVENT_ERROR, new SocketIOException("timeout"));
self.emit(EVENT_CONNECT_TIMEOUT, timeout); self.emit(EVENT_CONNECT_TIMEOUT, timeout);
} }
});
}
}, timeout, TimeUnit.MILLISECONDS); }, timeout, TimeUnit.MILLISECONDS);
On.Handle timeSub = new On.Handle() { On.Handle timeSub = new On.Handle() {
@@ -173,14 +186,15 @@ public class Manager extends Emitter {
} }
}; };
this.subs.add(timeSub); Manager.this.subs.add(timeSub);
} }
this.subs.add(openSub); Manager.this.subs.add(openSub);
this.subs.add(errorSub); Manager.this.subs.add(errorSub);
this.engine.open();
Manager.this.engine.open();
}
});
return this; return this;
} }
@@ -223,21 +237,24 @@ public class Manager extends Emitter {
Socket socket = this.nsps.get(nsp); Socket socket = this.nsps.get(nsp);
if (socket == null) { if (socket == null) {
socket = new Socket(this, nsp); socket = new Socket(this, nsp);
this.nsps.put(nsp, socket); Socket _socket = this.nsps.putIfAbsent(nsp, socket);
if (_socket != null) {
socket = _socket;
} else {
final Manager self = this; final Manager self = this;
socket.on(Socket.EVENT_CONNECT, new Listener() { socket.on(Socket.EVENT_CONNECT, new Listener() {
@Override @Override
public void call(Object... objects) { public void call(Object... objects) {
self.connected.incrementAndGet(); self.connected++;
} }
}); });
}
} }
return socket; return socket;
} }
/*package*/ void destroy(Socket socket) { /*package*/ void destroy(Socket socket) {
int connected = this.connected.decrementAndGet(); this.connected--;
if (connected == 0) { if (connected == 0) {
this.close(); this.close();
} }
@@ -269,18 +286,21 @@ public class Manager extends Emitter {
private void reconnect() { private void reconnect() {
final Manager self = this; final Manager self = this;
int attempts = this.attempts.incrementAndGet(); this.attempts++;
if (attempts > this._reconnectionAttempts) { if (attempts > this._reconnectionAttempts) {
this.emit(EVENT_RECONNECT_FAILED); this.emit(EVENT_RECONNECT_FAILED);
this.reconnecting = false; this.reconnecting = false;
} else { } else {
long delay = this.attempts.get() * this.reconnectionDelay(); long delay = this.attempts * this.reconnectionDelay();
delay = Math.min(delay, this.reconnectionDelayMax()); delay = Math.min(delay, this.reconnectionDelayMax());
logger.fine(String.format("will wait %dms before reconnect attempt", delay)); logger.fine(String.format("will wait %dms before reconnect attempt", delay));
this.reconnecting = true; this.reconnecting = true;
final Future timer = this.reconnectScheduler.schedule(new Runnable() { final Future timer = this.reconnectScheduler.schedule(new Runnable() {
@Override
public void run() {
EventThread.exec(new Runnable() {
@Override @Override
public void run() { public void run() {
logger.fine("attempting reconnect"); logger.fine("attempting reconnect");
@@ -298,6 +318,8 @@ public class Manager extends Emitter {
} }
}); });
} }
});
}
}, delay, TimeUnit.MILLISECONDS); }, delay, TimeUnit.MILLISECONDS);
this.subs.add(new On.Handle() { this.subs.add(new On.Handle() {
@@ -310,12 +332,13 @@ public class Manager extends Emitter {
} }
private void onreconnect() { private void onreconnect() {
int attempts = this.attempts.get(); int attempts = this.attempts;
this.attempts.set(0); this.attempts = 0;
this.reconnecting = false; this.reconnecting = false;
this.emit(EVENT_RECONNECT, attempts); this.emit(EVENT_RECONNECT, attempts);
} }
public static interface OpenCallback { public static interface OpenCallback {
public void call(Exception err); public void call(Exception err);

76
src/main/java/com/github/nkzawa/socketio/client/Socket.java
View File

@@ -1,6 +1,7 @@
package com.github.nkzawa.socketio.client; package com.github.nkzawa.socketio.client;
import com.github.nkzawa.emitter.Emitter; import com.github.nkzawa.emitter.Emitter;
import com.github.nkzawa.engineio.client.EventThread;
import com.github.nkzawa.socketio.parser.Packet; import com.github.nkzawa.socketio.parser.Packet;
import com.github.nkzawa.socketio.parser.Parser; import com.github.nkzawa.socketio.parser.Parser;
import com.google.gson.Gson; import com.google.gson.Gson;
@@ -8,9 +9,6 @@ import com.google.gson.JsonArray;
import com.google.gson.JsonElement; import com.google.gson.JsonElement;
import java.util.*; import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger; import java.util.logging.Logger;
public class Socket extends Emitter { public class Socket extends Emitter {
@@ -32,12 +30,13 @@ public class Socket extends Emitter {
private boolean connected; private boolean connected;
private boolean disconnected = true; private boolean disconnected = true;
private AtomicInteger ids = new AtomicInteger(); private int ids;
private String nsp; private String nsp;
private Manager io; private Manager io;
private Map<Integer, Ack> acks = new ConcurrentHashMap<Integer, Ack>(); private Map<Integer, Ack> acks = new HashMap<Integer, Ack>();
private Queue<On.Handle> subs; private Queue<On.Handle> subs;
private final Queue<LinkedList<Object>> buffer = new ConcurrentLinkedQueue<LinkedList<Object>>(); private final Queue<LinkedList<Object>> buffer = new LinkedList<LinkedList<Object>>();
public Socket(Manager io, String nsp) { public Socket(Manager io, String nsp) {
this.io = io; this.io = io;
@@ -45,8 +44,11 @@ public class Socket extends Emitter {
} }
public void open() { public void open() {
final Manager io = this.io; EventThread.exec(new Runnable() {
this.subs = new ConcurrentLinkedQueue<On.Handle>() {{ @Override
public void run() {
final Manager io = Socket.this.io;
Socket.this.subs = new LinkedList<On.Handle>() {{
add(On.on(io, Manager.EVENT_OPEN, new Listener() { add(On.on(io, Manager.EVENT_OPEN, new Listener() {
@Override @Override
public void call(Object... objects) { public void call(Object... objects) {
@@ -60,35 +62,47 @@ public class Socket extends Emitter {
} }
})); }));
}}; }};
if (this.io.readyState == Manager.OPEN) this.onopen(); if (Socket.this.io.readyState == Manager.OPEN) Socket.this.onopen();
io.open(); io.open();
} }
});
}
public void connect() { public void connect() {
this.open(); this.open();
} }
public Socket send(Object... args) { public Socket send(final Object... args) {
this.emit(EVENT_MESSAGE, args); EventThread.exec(new Runnable() {
@Override
public void run() {
Socket.this.emit(EVENT_MESSAGE, args);
}
});
return this; return this;
} }
@Override @Override
public Emitter emit(String event, Object... args) { public Emitter emit(final String event, final Object... args) {
EventThread.exec(new Runnable() {
@Override
public void run() {
if (events.contains(event)) { if (events.contains(event)) {
super.emit(event, args); Socket.super.emit(event, args);
} else { } else {
LinkedList<Object> _args = new LinkedList<Object>(Arrays.asList(args)); LinkedList<Object> _args = new LinkedList<Object>(Arrays.asList(args));
if (_args.peekLast() instanceof Ack) { if (_args.peekLast() instanceof Ack) {
Ack ack = (Ack)_args.pollLast(); Ack ack = (Ack)_args.pollLast();
return this.emit(event, _args.toArray(), ack); Socket.this.emit(event, _args.toArray(), ack);
return;
} }
_args.offerFirst(event); _args.offerFirst(event);
Packet packet = new Packet(Parser.EVENT, toJsonArray(_args)); Packet packet = new Packet(Parser.EVENT, toJsonArray(_args));
this.packet(packet); Socket.this.packet(packet);
} }
}
});
return this; return this;
} }
@@ -100,20 +114,23 @@ public class Socket extends Emitter {
* @param ack * @param ack
* @return * @return
*/ */
public Emitter emit(final String event, final Object[] args, Ack ack) { public Emitter emit(final String event, final Object[] args, final Ack ack) {
EventThread.exec(new Runnable() {
@Override
public void run() {
List<Object> _args = new ArrayList<Object>() {{ List<Object> _args = new ArrayList<Object>() {{
add(event); add(event);
addAll(Arrays.asList(args)); addAll(Arrays.asList(args));
}}; }};
Packet packet = new Packet(Parser.EVENT, toJsonArray(_args)); Packet packet = new Packet(Parser.EVENT, toJsonArray(_args));
int ids = this.ids.getAndIncrement();
logger.fine(String.format("emitting packet with ack id %d", ids)); logger.fine(String.format("emitting packet with ack id %d", ids));
this.acks.put(ids, ack); Socket.this.acks.put(ids, ack);
packet.id = ids; packet.id = ids++;
this.packet(packet);
Socket.this.packet(packet);
}
});
return this; return this;
} }
@@ -255,14 +272,19 @@ public class Socket extends Emitter {
} }
public Socket close() { public Socket close() {
if (!this.connected) return this; EventThread.exec(new Runnable() {
@Override
public void run() {
if (!Socket.this.connected) return;
logger.fine(String.format("performing disconnect (%s)", this.nsp)); logger.fine(String.format("performing disconnect (%s)", Socket.this.nsp));
this.packet(new Packet(Parser.DISCONNECT)); Socket.this.packet(new Packet(Parser.DISCONNECT));
this.destroy(); Socket.this.destroy();
this.onclose("io client disconnect"); Socket.this.onclose("io client disconnect");
}
});
return this; return this;
} }