diff --git a/core/src/main/java/com/glines/socketio/server/DefaultSession.java b/core/src/main/java/com/glines/socketio/server/DefaultSession.java index 9bd639f..2c87514 100644 --- a/core/src/main/java/com/glines/socketio/server/DefaultSession.java +++ b/core/src/main/java/com/glines/socketio/server/DefaultSession.java @@ -27,7 +27,11 @@ import com.glines.socketio.common.ConnectionState; import com.glines.socketio.common.DisconnectReason; import com.glines.socketio.common.SocketIOException; +import com.glines.socketio.util.JSON; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicLong; @@ -135,7 +139,7 @@ private void sendHeartBeat() { LOGGER.log(Level.FINE, "handler.sendMessage failed: ", e); handler.abort(); } - startTimeoutTimer(); + startHeartbeatTimer(); } @Override @@ -212,6 +216,9 @@ public void onMessage(SocketIOFrame message) { LOGGER.log(Level.FINE, "Session[" + sessionId + "]: onMessage: " + message.getData()); onMessage(message.getData()); break; + case EVENT: + this.dispatchEvent(message); + break; default: // Ignore unknown message types break; @@ -281,7 +288,7 @@ public void onConnect(TransportHandler handler) { if (handler == null) { state = ConnectionState.CLOSED; inbound = null; - socketIOSessionManager.socketIOSessions.remove(sessionId); + socketIOSessionManager.removeSession(sessionId); } else if (this.handler == null) { this.handler = handler; if (inbound == null) { @@ -316,10 +323,31 @@ public void onMessage(String message) { } } + public void dispatchEvent(SocketIOFrame message){ + if (inbound != null) { + try { + Map event = (Map) JSON.parse(message.getData()); + String eventName = (String) event.get("name"); + Object[] objects = (Object[]) event.get("args"); + String[] args = new String[objects.length]; + for(int i = 0; i < objects.length; i++) { + args[i] = JSON.toString(objects[i]); + } + Method callback = inbound.getClass().getDeclaredMethod("on" + eventName.substring(0, 1).toUpperCase() + eventName.substring(1), new Class[]{ String[].class }); + callback.setAccessible(true); + callback.invoke(inbound, new Object[]{ args }); + } catch (Throwable e) { + if (LOGGER.isLoggable(Level.WARNING)) + LOGGER.log(Level.WARNING, "Session[" + sessionId + "]: Exception thrown by SocketIOInbound.onMessage()", e); + } + } + } + @Override public void onDisconnect(DisconnectReason reason) { if (LOGGER.isLoggable(Level.FINE)) LOGGER.log(Level.FINE, "Session[" + sessionId + "]: onDisconnect: " + reason); + clearTimeoutTimer(); clearHeartbeatTimer(); if (inbound != null) { @@ -349,6 +377,6 @@ public void onShutdown() { onDisconnect(DisconnectReason.ERROR); } } - socketIOSessionManager.socketIOSessions.remove(sessionId); + socketIOSessionManager.removeSession(sessionId); } } diff --git a/core/src/main/java/com/glines/socketio/server/SocketIOServlet.java b/core/src/main/java/com/glines/socketio/server/SocketIOServlet.java index b4727c3..e148049 100644 --- a/core/src/main/java/com/glines/socketio/server/SocketIOServlet.java +++ b/core/src/main/java/com/glines/socketio/server/SocketIOServlet.java @@ -133,7 +133,7 @@ private void serve(HttpServletRequest request, HttpServletResponse response) thr OutputStream os = response.getOutputStream(); // Format: sessionId : heartbeat : timeout - String body = request.getSession().getId().toString() + + String body = SocketIOSessionManager.generateSessionId() + ":" + (config.getString(DEFAULT_HEARTBEAT_TIMEOUT) == null ? "15000" : config.getString(DEFAULT_HEARTBEAT_TIMEOUT)) + ":" + (config.getString(DEFAULT_TIMEOUT) == null? "10000" : config.getString(DEFAULT_TIMEOUT)) + ":"; diff --git a/core/src/main/java/com/glines/socketio/server/SocketIOSessionManager.java b/core/src/main/java/com/glines/socketio/server/SocketIOSessionManager.java index 37ce4ce..5d72b35 100644 --- a/core/src/main/java/com/glines/socketio/server/SocketIOSessionManager.java +++ b/core/src/main/java/com/glines/socketio/server/SocketIOSessionManager.java @@ -24,14 +24,36 @@ */ package com.glines.socketio.server; +import java.security.SecureRandom; +import java.util.Random; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; public final class SocketIOSessionManager implements SessionManager { + private static final char[] BASE64_ALPHABET = + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_" + .toCharArray(); + private static final int SESSION_ID_LENGTH = 20; - final ConcurrentMap socketIOSessions = new ConcurrentHashMap(); + private static Random random = new SecureRandom(); + + private static String generateRandomString(int length) { + StringBuilder result = new StringBuilder(length); + byte[] bytes = new byte[length]; + random.nextBytes(bytes); + for (int i = 0; i < bytes.length; i++) { + result.append(BASE64_ALPHABET[bytes[i] & 0x3F]); + } + return result.toString(); + } + + public static String generateSessionId() { + return generateRandomString(SESSION_ID_LENGTH); + } + + final private ConcurrentMap socketIOSessions = new ConcurrentHashMap(); final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); @Override @@ -45,4 +67,8 @@ public SocketIOSession createSession(SocketIOInbound inbound, String sessionId) public SocketIOSession getSession(String sessionId) { return socketIOSessions.get(sessionId); } + + public void removeSession(String sessionId) { + this.socketIOSessions.remove(sessionId); + } } diff --git a/core/src/main/java/com/glines/socketio/server/transport/XHRPollingDataHandler.java b/core/src/main/java/com/glines/socketio/server/transport/XHRPollingDataHandler.java index a457dd9..8a82c42 100644 --- a/core/src/main/java/com/glines/socketio/server/transport/XHRPollingDataHandler.java +++ b/core/src/main/java/com/glines/socketio/server/transport/XHRPollingDataHandler.java @@ -73,7 +73,7 @@ public void onStartSend(HttpServletResponse response) throws IOException { @Override public void onWriteData(ServletResponse response, String data) throws IOException { - response.getOutputStream().print(data); + response.getOutputStream().write(data.getBytes()); response.flushBuffer(); } diff --git a/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/JettyContinuationTransportHandler.java b/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/JettyContinuationTransportHandler.java index ae5bb9d..39327ea 100644 --- a/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/JettyContinuationTransportHandler.java +++ b/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/JettyContinuationTransportHandler.java @@ -160,7 +160,7 @@ public void sendMessage(SocketIOFrame frame) throws SocketIOException { } } else { String data = frame.encode(); - if (!buffer.putMessage(data, maxIdleTime)) { + if (!buffer.putMessage(data)) { getSession().onDisconnect(DisconnectReason.TIMEOUT); abort(); throw new SocketIOException(); @@ -200,14 +200,10 @@ public void handle(HttpServletRequest request, HttpServletResponse response, Soc } else { if (!dataHandler.isConnectionPersistent()) { if (!buffer.isEmpty()) { - List messages = buffer.drainMessages(); - if (messages.size() > 0) { - StringBuilder data = new StringBuilder(); - for (String msg : messages) { - data.append(msg); - } + String msg = buffer.getMessage(maxIdleTime); + if (msg != null) { dataHandler.onStartSend(response); - dataHandler.onWriteData(response, data.toString()); + dataHandler.onWriteData(response, msg); dataHandler.onFinishSend(response); if (!disconnectWhenEmpty) { getSession().startTimeoutTimer(); @@ -302,6 +298,8 @@ public void onComplete(Continuation cont) { } } } + }else{ + getSession().startTimeoutTimer(); } } @@ -320,6 +318,7 @@ public void onTimeout(Continuation cont) { } else { try { dataHandler.onFinishSend(cont.getServletResponse()); + cont.complete(); } catch (IOException e) { getSession().onDisconnect(DisconnectReason.DISCONNECT); abort(); diff --git a/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/TransportBuffer.java b/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/TransportBuffer.java index c433b5e..c58af1a 100644 --- a/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/TransportBuffer.java +++ b/extension/jetty/src/main/java/com/glines/socketio/server/transport/jetty/TransportBuffer.java @@ -104,7 +104,7 @@ public String getMessage(long timeout) { } } - public boolean putMessage(String message, long timeout) { + public boolean putMessage(String message) { BufferListener listener = listenerRef.get(); if (listener != null) { try { @@ -120,15 +120,11 @@ public boolean putMessage(String message, long timeout) { return false; } } else { - try { - if (!inputSemaphore.tryAcquire(message.length(), timeout, TimeUnit.MILLISECONDS)) { - return false; - } - queue.offer(message); - return true; - } catch (InterruptedException e) { + if (!inputSemaphore.tryAcquire(message.length())) { return false; } + queue.offer(message); + return true; } } }