diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractHttpSockJsSession.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractHttpSockJsSession.java index bd3dccd613..c2f8de4718 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractHttpSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractHttpSockJsSession.java @@ -30,7 +30,6 @@ import org.springframework.http.HttpHeaders; import org.springframework.http.server.ServerHttpAsyncRequestControl; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.util.Assert; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.WebSocketExtension; import org.springframework.web.socket.WebSocketHandler; @@ -41,7 +40,7 @@ import org.springframework.web.socket.sockjs.frame.SockJsFrameFormat; import org.springframework.web.socket.sockjs.transport.SockJsServiceConfig; /** - * An abstract base class for use with HTTP transport based SockJS sessions. + * An abstract base class for use with HTTP transport SockJS sessions. * * @author Rossen Stoyanchev * @since 4.0 @@ -64,9 +63,12 @@ public abstract class AbstractHttpSockJsSession extends AbstractSockJsSession { private volatile ServerHttpResponse response; + private volatile SockJsFrameFormat frameFormat; + + private volatile ServerHttpAsyncRequestControl asyncRequestControl; - private volatile SockJsFrameFormat frameFormat; + private final Object responseLock = new Object(); private volatile boolean requestInitialized; @@ -124,18 +126,10 @@ public abstract class AbstractHttpSockJsSession extends AbstractSockJsSession { return this.acceptedProtocol; } - /** - * Return response for the current request, or {@code null} if between requests. - */ - protected ServerHttpResponse getResponse() { - return this.response; - } - /** * Return the SockJS buffer for messages stored transparently between polling * requests. If the polling request takes longer than 5 seconds, the session - * will be closed. - * + * is closed. * @see org.springframework.web.socket.sockjs.transport.TransportHandlingSockJsService */ protected Queue getMessageCache() { @@ -173,123 +167,104 @@ public abstract class AbstractHttpSockJsSession extends AbstractSockJsSession { return Collections.emptyList(); } + /** + * Whether this HTTP transport streams message frames vs closing the response + * after each frame written (long polling). + */ + protected abstract boolean isStreaming(); + /** - * Handle the first HTTP request, i.e. the one that starts a SockJS session. - * Write a prelude to the response (if needed), send the SockJS "open" frame - * to indicate to the client the session is opened, and invoke the - * delegate WebSocketHandler to provide it with the newly opened session. - *

- * The "xhr" and "jsonp" (polling-based) transports completes the initial request - * as soon as the open frame is sent. Following that the client should start a - * successive polling request within the same SockJS session. - *

- * The "xhr_streaming", "eventsource", and "htmlfile" transports are streaming - * based and will leave the initial request open in order to stream one or - * more messages. However, even streaming based transports eventually recycle - * the long running request, after a certain number of bytes have been streamed - * (128K by default), and allow the client to start a successive request within - * the same SockJS session. + * Handle the first request for receiving messages on a SockJS HTTP transport + * based session. + * + *

Long polling-based transports (e.g. "xhr", "jsonp") complete the request + * after writing the open frame. Streaming-based transports ("xhr_streaming", + * "eventsource", and "htmlfile") leave the response open longer for further + * streaming of message frames but will also close it eventually after some + * amount of data has been sent. * * @param request the current request * @param response the current response * @param frameFormat the transport-specific SocksJS frame format to use - * - * @see #handleSuccessiveRequest(org.springframework.http.server.ServerHttpRequest, org.springframework.http.server.ServerHttpResponse, org.springframework.web.socket.sockjs.frame.SockJsFrameFormat) */ public void handleInitialRequest(ServerHttpRequest request, ServerHttpResponse response, SockJsFrameFormat frameFormat) throws SockJsException { - initRequest(request, response, frameFormat); - this.uri = request.getURI(); this.handshakeHeaders = request.getHeaders(); this.principal = request.getPrincipal(); this.localAddress = request.getLocalAddress(); this.remoteAddress = request.getRemoteAddress(); - try { - writePrelude(request, response); - writeFrame(SockJsFrame.openFrame()); - } - catch (Throwable ex) { - tryCloseWithSockJsTransportError(ex, CloseStatus.SERVER_ERROR); - throw new SockJsTransportFailureException("Failed to send \"open\" frame", getId(), ex); - } - - try { - this.requestInitialized = true; - delegateConnectionEstablished(); - } - catch (Throwable ex) { - throw new SockJsException("Unhandled exception from WebSocketHandler", getId(), ex); - } - } - - private void initRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsFrameFormat frameFormat) { - - Assert.notNull(request, "Request must not be null"); - Assert.notNull(response, "Response must not be null"); - Assert.notNull(frameFormat, "SockJsFrameFormat must not be null"); - this.response = response; this.frameFormat = frameFormat; this.asyncRequestControl = request.getAsyncRequestControl(response); + + try { + // Let "our" handler know before sending the open frame to the remote handler + delegateConnectionEstablished(); + writePrelude(request, response); + writeFrame(SockJsFrame.openFrame()); + if (isStreaming() && !isClosed()) { + startAsyncRequest(); + } + } + catch (Throwable ex) { + tryCloseWithSockJsTransportError(ex, CloseStatus.SERVER_ERROR); + throw new SockJsTransportFailureException("Failed to open session", getId(), ex); + } } protected void writePrelude(ServerHttpRequest request, ServerHttpResponse response) throws IOException { } + private void startAsyncRequest() { + this.asyncRequestControl.start(-1); + if (this.messageCache.size() > 0) { + flushCache(); + } + else { + scheduleHeartbeat(); + } + this.requestInitialized = true; + } + /** - * Handle all HTTP requests part of the same SockJS session except for the very - * first, initial request. Write a prelude (if needed) and keep the request - * open and ready to send a message from the server to the client. - *

- * The "xhr" and "jsonp" (polling-based) transports completes the request when - * the next message is sent, which could be an array of messages cached during - * the time between successive requests, or it could be a heartbeat message - * sent if no other messages were sent (by default within 25 seconds). - *

- * The "xhr_streaming", "eventsource", and "htmlfile" transports are streaming - * based and will leave the request open longer in order to stream messages over - * a period of time. However, even streaming based transports eventually recycle - * the long running request, after a certain number of bytes have been streamed - * (128K by default), and allow the client to start a successive request within - * the same SockJS session. + * Handle all requests, except the first one, to receive messages on a SockJS + * HTTP transport based session. + * + *

Long polling-based transports (e.g. "xhr", "jsonp") complete the request + * after writing any buffered message frames (or the next one). Streaming-based + * transports ("xhr_streaming", "eventsource", and "htmlfile") leave the + * response open longer for further streaming of message frames but will also + * close it eventually after some amount of data has been sent. * * @param request the current request * @param response the current response * @param frameFormat the transport-specific SocksJS frame format to use - * - * @see #handleInitialRequest(org.springframework.http.server.ServerHttpRequest, org.springframework.http.server.ServerHttpResponse, org.springframework.web.socket.sockjs.frame.SockJsFrameFormat) */ public void handleSuccessiveRequest(ServerHttpRequest request, ServerHttpResponse response, SockJsFrameFormat frameFormat) throws SockJsException { - initRequest(request, response, frameFormat); - try { - writePrelude(request, response); + synchronized (this.responseLock) { + try { + if (isClosed()) { + response.getBody().write(SockJsFrame.closeFrameGoAway().getContentBytes()); + } + this.response = response; + this.frameFormat = frameFormat; + this.asyncRequestControl = request.getAsyncRequestControl(response); + writePrelude(request, response); + startAsyncRequest(); + } + catch (Throwable ex) { + tryCloseWithSockJsTransportError(ex, CloseStatus.SERVER_ERROR); + throw new SockJsTransportFailureException("Failed to handle SockJS receive request", getId(), ex); + } } - catch (Throwable ex) { - tryCloseWithSockJsTransportError(ex, CloseStatus.SERVER_ERROR); - throw new SockJsTransportFailureException("Failed to send \"open\" frame", getId(), ex); - } - startAsyncRequest(); } - protected void startAsyncRequest() throws SockJsException { - try { - this.asyncRequestControl.start(-1); - this.requestInitialized = true; - scheduleHeartbeat(); - tryFlushCache(); - } - catch (Throwable ex) { - tryCloseWithSockJsTransportError(ex, CloseStatus.SERVER_ERROR); - throw new SockJsTransportFailureException("Failed to flush messages", getId(), ex); - } - } @Override protected final void sendMessageInternal(String message) throws SockJsTransportFailureException { @@ -297,27 +272,35 @@ public abstract class AbstractHttpSockJsSession extends AbstractSockJsSession { tryFlushCache(); } - private void tryFlushCache() throws SockJsTransportFailureException { - if (this.messageCache.isEmpty()) { - logger.trace("Nothing to flush"); - return; - } - if (logger.isTraceEnabled()) { - logger.trace(this.messageCache.size() + " message(s) to flush"); - } - if (isActive() && this.requestInitialized) { - logger.trace("Flushing messages"); - flushCache(); - } - else { + private boolean tryFlushCache() throws SockJsTransportFailureException { + synchronized (this.responseLock) { + if (this.messageCache.isEmpty()) { + logger.trace("Nothing to flush in session=" + this.getId()); + return false; + } if (logger.isTraceEnabled()) { - logger.trace("Not ready to flush"); + logger.trace(this.messageCache.size() + " message(s) to flush in session " + this.getId()); + } + if (isActive() && this.requestInitialized) { + if (logger.isTraceEnabled()) { + logger.trace("Session is active, ready to flush."); + } + cancelHeartbeat(); + flushCache(); + return true; + } + else { + if (logger.isTraceEnabled()) { + logger.trace("Session is not active, not ready to flush."); + } + return false; } } } /** - * Only called if the connection is currently active + * Called when the connection is active and ready to write to the response. + * Sub-classes should implement but never call this method directly. */ protected abstract void flushCache() throws SockJsTransportFailureException; @@ -327,25 +310,27 @@ public abstract class AbstractHttpSockJsSession extends AbstractSockJsSession { } protected void resetRequest() { + synchronized (this.responseLock) { - this.requestInitialized = false; - updateLastActiveTime(); - - if (isActive()) { ServerHttpAsyncRequestControl control = this.asyncRequestControl; - if (control.isStarted()) { - try { - logger.debug("Completing asynchronous request"); - control.complete(); - } - catch (Throwable ex) { - logger.error("Failed to complete request: " + ex.getMessage()); + this.asyncRequestControl = null; + this.requestInitialized = false; + this.response = null; + + updateLastActiveTime(); + + if (control != null && !control.isCompleted()) { + if (control.isStarted()) { + try { + logger.debug("Completing asynchronous request"); + control.complete(); + } + catch (Throwable ex) { + logger.error("Failed to complete request: " + ex.getMessage()); + } } } } - - this.response = null; - this.asyncRequestControl = null; } @Override @@ -353,9 +338,15 @@ public abstract class AbstractHttpSockJsSession extends AbstractSockJsSession { if (isActive()) { String formattedFrame = this.frameFormat.format(frame); if (logger.isTraceEnabled()) { - logger.trace("Writing " + formattedFrame); + logger.trace("Writing to HTTP response: " + formattedFrame); + } + this.response.getBody().write(formattedFrame.getBytes(SockJsFrame.CHARSET)); + if (isStreaming()) { + this.response.flush(); + } + else { + resetRequest(); } - getResponse().getBody().write(formattedFrame.getBytes(SockJsFrame.CHARSET)); } } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractSockJsSession.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractSockJsSession.java index 215a69f6c3..0a0ec5770a 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractSockJsSession.java @@ -40,6 +40,7 @@ import org.springframework.web.socket.WebSocketMessage; import org.springframework.web.socket.sockjs.SockJsMessageDeliveryException; import org.springframework.web.socket.sockjs.SockJsTransportFailureException; import org.springframework.web.socket.sockjs.frame.SockJsFrame; +import org.springframework.web.socket.sockjs.frame.SockJsMessageCodec; import org.springframework.web.socket.sockjs.transport.SockJsServiceConfig; import org.springframework.web.socket.sockjs.transport.SockJsSession; @@ -142,6 +143,10 @@ public abstract class AbstractSockJsSession implements SockJsSession { return this.id; } + protected SockJsMessageCodec getMessageCodec() { + return this.config.getMessageCodec(); + } + public SockJsServiceConfig getSockJsServiceConfig() { return this.config; } @@ -420,7 +425,9 @@ public abstract class AbstractSockJsSession implements SockJsSession { @Override public String toString() { - return "SockJS session id=" + this.id; + long currentTime = System.currentTimeMillis(); + return "SockJsSession[id=" + this.id + ", state=" + this.state + ", sinceCreated=" + + (currentTime - this.timeCreated) + ", sinceLastActive=" + (currentTime - this.timeLastActive) + "]"; } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/PollingSockJsSession.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/PollingSockJsSession.java index 52e27829a6..fa2f721722 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/PollingSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/PollingSockJsSession.java @@ -16,6 +16,8 @@ package org.springframework.web.socket.sockjs.transport.session; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.Queue; @@ -33,6 +35,7 @@ import org.springframework.web.socket.sockjs.transport.SockJsServiceConfig; */ public class PollingSockJsSession extends AbstractHttpSockJsSession { + public PollingSockJsSession(String sessionId, SockJsServiceConfig config, WebSocketHandler wsHandler, Map attributes) { @@ -41,22 +44,20 @@ public class PollingSockJsSession extends AbstractHttpSockJsSession { @Override - protected void flushCache() throws SockJsTransportFailureException { - cancelHeartbeat(); - Queue messageCache = getMessageCache(); - String[] messages = messageCache.toArray(new String[messageCache.size()]); - messageCache.clear(); + protected boolean isStreaming() { + return false; + } + @Override + protected void flushCache() throws SockJsTransportFailureException { + String[] messages = new String[getMessageCache().size()]; + for (int i = 0; i < messages.length; i++) { + messages[i] = getMessageCache().poll(); + } SockJsMessageCodec messageCodec = getSockJsServiceConfig().getMessageCodec(); SockJsFrame frame = SockJsFrame.messageFrame(messageCodec, messages); writeFrame(frame); } - @Override - protected void writeFrame(SockJsFrame frame) throws SockJsTransportFailureException { - super.writeFrame(frame); - resetRequest(); - } - } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/StreamingSockJsSession.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/StreamingSockJsSession.java index 70facc8776..1e2d301dbe 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/StreamingSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/StreamingSockJsSession.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,7 +16,6 @@ package org.springframework.web.socket.sockjs.transport.session; -import java.io.IOException; import java.util.Map; import org.springframework.http.server.ServerHttpRequest; @@ -48,21 +47,12 @@ public class StreamingSockJsSession extends AbstractHttpSockJsSession { @Override - public void handleInitialRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsFrameFormat frameFormat) throws SockJsException { - - super.handleInitialRequest(request, response, frameFormat); - - // the WebSocketHandler delegate may have closed the session - if (!isClosed()) { - super.startAsyncRequest(); - } + protected boolean isStreaming() { + return true; } @Override protected void flushCache() throws SockJsTransportFailureException { - cancelHeartbeat(); - do { String message = getMessageCache().poll(); SockJsMessageCodec messageCodec = getSockJsServiceConfig().getMessageCodec(); @@ -79,26 +69,12 @@ public class StreamingSockJsSession extends AbstractHttpSockJsSession { logger.trace("Streamed bytes limit reached. Recycling current request"); } resetRequest(); + this.byteCount = 0; break; } } while (!getMessageCache().isEmpty()); - scheduleHeartbeat(); } - @Override - protected void resetRequest() { - super.resetRequest(); - this.byteCount = 0; - } - - @Override - protected void writeFrameInternal(SockJsFrame frame) throws IOException { - if (isActive()) { - super.writeFrameInternal(frame); - getResponse().flush(); - } - } - } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/WebSocketServerSockJsSession.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/WebSocketServerSockJsSession.java index 51e2df3f22..d250d9e34c 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/WebSocketServerSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/WebSocketServerSockJsSession.java @@ -22,6 +22,10 @@ import java.net.URI; import java.security.Principal; import java.util.List; import java.util.Map; +import java.util.Queue; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.springframework.http.HttpHeaders; import org.springframework.util.Assert; @@ -34,7 +38,6 @@ import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.adapter.NativeWebSocketSession; import org.springframework.web.socket.sockjs.SockJsTransportFailureException; import org.springframework.web.socket.sockjs.frame.SockJsFrame; -import org.springframework.web.socket.sockjs.frame.SockJsMessageCodec; import org.springframework.web.socket.sockjs.transport.SockJsServiceConfig; /** @@ -47,6 +50,12 @@ public class WebSocketServerSockJsSession extends AbstractSockJsSession implemen private WebSocketSession webSocketSession; + private volatile boolean openFrameSent; + + private final Queue initSessionCache = new LinkedBlockingDeque(); + + private final Lock initSessionLock = new ReentrantLock(); + public WebSocketServerSockJsSession(String id, SockJsServiceConfig config, WebSocketHandler handler, Map attributes) { @@ -143,15 +152,23 @@ public class WebSocketServerSockJsSession extends AbstractSockJsSession implemen public void initializeDelegateSession(WebSocketSession session) { - this.webSocketSession = session; - try { - TextMessage message = new TextMessage(SockJsFrame.openFrame().getContent()); - this.webSocketSession.sendMessage(message); - scheduleHeartbeat(); - delegateConnectionEstablished(); - } - catch (Exception ex) { - tryCloseWithSockJsTransportError(ex, CloseStatus.SERVER_ERROR); + synchronized (this.initSessionLock) { + this.webSocketSession = session; + try { + // Let "our" handler know before sending the open frame to the remote handler + delegateConnectionEstablished(); + this.webSocketSession.sendMessage(new TextMessage(SockJsFrame.openFrame().getContent())); + + // Flush any messages cached in the mean time + while (!this.initSessionCache.isEmpty()) { + writeFrame(SockJsFrame.messageFrame(getMessageCodec(), this.initSessionCache.poll())); + } + scheduleHeartbeat(); + this.openFrameSent = true; + } + catch (Exception ex) { + tryCloseWithSockJsTransportError(ex, CloseStatus.SERVER_ERROR); + } } } @@ -180,10 +197,20 @@ public class WebSocketServerSockJsSession extends AbstractSockJsSession implemen @Override public void sendMessageInternal(String message) throws SockJsTransportFailureException { + + // Open frame not sent yet? + // If in the session initialization thread, then cache, otherwise wait. + + if (!this.openFrameSent) { + synchronized (this.initSessionLock) { + if (!this.openFrameSent) { + this.initSessionCache.add(message); + return; + } + } + } cancelHeartbeat(); - SockJsMessageCodec messageCodec = getSockJsServiceConfig().getMessageCodec(); - SockJsFrame frame = SockJsFrame.messageFrame(messageCodec, message); - writeFrame(frame); + writeFrame(SockJsFrame.messageFrame(getMessageCodec(), message)); scheduleHeartbeat(); } diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/client/JettySockJsIntegrationTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/client/JettySockJsIntegrationTests.java index 89120794ec..31e1868b30 100644 --- a/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/client/JettySockJsIntegrationTests.java +++ b/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/client/JettySockJsIntegrationTests.java @@ -27,9 +27,6 @@ import org.springframework.web.socket.client.jetty.JettyWebSocketClient; import org.springframework.web.socket.server.RequestUpgradeStrategy; import org.springframework.web.socket.server.jetty.JettyRequestUpgradeStrategy; -import java.util.ArrayList; -import java.util.List; - /** * SockJS integration tests using Jetty for client and server. * diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/transport/session/HttpSockJsSessionTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/transport/session/HttpSockJsSessionTests.java index d0c96bdf4c..b2ec62daea 100644 --- a/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/transport/session/HttpSockJsSessionTests.java +++ b/spring-websocket/src/test/java/org/springframework/web/socket/sockjs/transport/session/HttpSockJsSessionTests.java @@ -94,10 +94,8 @@ public class HttpSockJsSessionTests extends AbstractSockJsSessionTests expected = Arrays.asList(new TextMessage("o"), new TextMessage("a[\"go go\"]")); + assertEquals(expected, this.webSocketSession.getSentMessages()); + } + @Test public void handleMessageEmptyPayload() throws Exception { this.session.handleMessage(new TextMessage(""), this.webSocketSession);