From e56559fd4de4d2c105c70c2f8e03cf92506b2510 Mon Sep 17 00:00:00 2001 From: Juergen Hoeller Date: Tue, 7 Oct 2014 13:45:07 +0200 Subject: [PATCH] WebSocketSession extends java.io.Closeable, plus reorganization of AbstractSockJsSession's code Issue: SPR-12311 --- .../web/socket/WebSocketSession.java | 21 +- .../session/AbstractSockJsSession.java | 357 +++++++++--------- 2 files changed, 193 insertions(+), 185 deletions(-) diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/WebSocketSession.java b/spring-websocket/src/main/java/org/springframework/web/socket/WebSocketSession.java index 6130fe925c..9a2d06b11d 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/WebSocketSession.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/WebSocketSession.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,6 +16,7 @@ package org.springframework.web.socket; +import java.io.Closeable; import java.io.IOException; import java.net.InetSocketAddress; import java.net.URI; @@ -32,7 +33,7 @@ import org.springframework.http.HttpHeaders; * @author Rossen Stoyanchev * @since 4.0 */ -public interface WebSocketSession { +public interface WebSocketSession extends Closeable { /** * Return a unique session identifier. @@ -51,12 +52,10 @@ public interface WebSocketSession { /** * Return the map with attributes associated with the WebSocket session. - * *

When the WebSocketSession is created, on the server side, the map can be * through a {@link org.springframework.web.socket.server.HandshakeInterceptor}. * On the client side, the map can be populated by passing attributes to the - * {@link org.springframework.web.socket.client.WebSocketClient} handshake - * methods. + * {@link org.springframework.web.socket.client.WebSocketClient} handshake methods. */ Map getAttributes(); @@ -109,23 +108,23 @@ public interface WebSocketSession { */ List getExtensions(); + /** + * Send a WebSocket message: either {@link TextMessage} or {@link BinaryMessage}. + */ + void sendMessage(WebSocketMessage message) throws IOException; + /** * Return whether the connection is still open. */ boolean isOpen(); - /** - * Send a WebSocket message either {@link TextMessage} or - * {@link BinaryMessage}. - */ - void sendMessage(WebSocketMessage message) throws IOException; - /** * Close the WebSocket connection with status 1000, i.e. equivalent to: * 

 	 * session.close(CloseStatus.NORMAL);
 	 *
*/ + @Override void close() throws IOException; /** 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 5495bfdf41..da3adeac65 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 @@ -53,21 +53,18 @@ import org.springframework.web.socket.sockjs.transport.SockJsSession; */ public abstract class AbstractSockJsSession implements SockJsSession { - protected final Log logger = LogFactory.getLog(getClass()); + private static enum State {NEW, OPEN, CLOSED} /** * Log category to use on network IO exceptions after a client has gone away. - * *

The Servlet API does not provide notifications when a client disconnects; * see SERVLET_SPEC-44. * Therefore network IO failures may occur simply because a client has gone away, * and that can fill the logs with unnecessary stack traces. - * *

We make a best effort to identify such network failures, on a per-server * basis, and log them under a separate log category. A simple one-line message * is logged at DEBUG level, while a full stack trace is shown at TRACE level. - * * @see #disconnectedClientLogger */ public static final String DISCONNECTED_CLIENT_LOG_CATEGORY = @@ -79,6 +76,7 @@ public abstract class AbstractSockJsSession implements SockJsSession { */ protected static final Log disconnectedClientLogger = LogFactory.getLog(DISCONNECTED_CLIENT_LOG_CATEGORY); + private static final Set disconnectedClientExceptions; static { @@ -91,6 +89,8 @@ public abstract class AbstractSockJsSession implements SockJsSession { } + protected final Log logger = LogFactory.getLog(getClass()); + private final String id; private final SockJsServiceConfig config; @@ -99,15 +99,12 @@ public abstract class AbstractSockJsSession implements SockJsSession { private final Map attributes = new ConcurrentHashMap(); - private volatile State state = State.NEW; - private final long timeCreated = System.currentTimeMillis(); private volatile long timeLastActive = this.timeCreated; - private volatile ScheduledFuture heartbeatTask; private volatile boolean heartbeatDisabled; @@ -115,7 +112,6 @@ public abstract class AbstractSockJsSession implements SockJsSession { /** * Create a new instance. - * * @param id the session ID * @param config SockJS service configuration options * @param handler the recipient of SockJS messages @@ -157,6 +153,22 @@ public abstract class AbstractSockJsSession implements SockJsSession { return this.attributes; } + + // Message sending + + public final void sendMessage(WebSocketMessage message) throws IOException { + Assert.state(!isClosed(), "Cannot send a message when session is closed"); + if (!(message instanceof TextMessage)) { + throw new IllegalArgumentException("Expected text message: " + message); + } + sendMessageInternal(((TextMessage) message).getPayload()); + } + + protected abstract void sendMessageInternal(String message) throws IOException; + + + // Lifecycle related methods + public boolean isNew() { return State.NEW.equals(this.state); } @@ -171,91 +183,7 @@ public abstract class AbstractSockJsSession implements SockJsSession { } /** - * Polling and Streaming sessions periodically close the current HTTP request and - * wait for the next request to come through. During this "downtime" the session is - * still open but inactive and unable to send messages and therefore has to buffer - * them temporarily. A WebSocket session by contrast is stateful and remain active - * until closed. - */ - public abstract boolean isActive(); - - @Override - public long getTimeSinceLastActive() { - if (isNew()) { - return (System.currentTimeMillis() - this.timeCreated); - } - else { - return isActive() ? 0 : System.currentTimeMillis() - this.timeLastActive; - } - } - - /** - * Should be invoked whenever the session becomes inactive. - */ - protected void updateLastActiveTime() { - this.timeLastActive = System.currentTimeMillis(); - } - - @Override - public void disableHeartbeat() { - this.heartbeatDisabled = true; - cancelHeartbeat(); - } - - public void delegateConnectionEstablished() throws Exception { - this.state = State.OPEN; - this.handler.afterConnectionEstablished(this); - } - - public void delegateMessages(String[] messages) throws SockJsMessageDeliveryException { - List undelivered = new ArrayList(Arrays.asList(messages)); - for (String message : messages) { - try { - if (isClosed()) { - throw new SockJsMessageDeliveryException(this.id, undelivered, "Session closed"); - } - else { - this.handler.handleMessage(this, new TextMessage(message)); - undelivered.remove(0); - } - } - catch (Throwable ex) { - throw new SockJsMessageDeliveryException(this.id, undelivered, ex); - } - } - } - - /** - * Invoked when the underlying connection is closed. - */ - public final void delegateConnectionClosed(CloseStatus status) throws Exception { - if (!isClosed()) { - try { - updateLastActiveTime(); - cancelHeartbeat(); - } - finally { - this.state = State.CLOSED; - this.handler.afterConnectionClosed(this, status); - } - } - } - - public void delegateError(Throwable ex) throws Exception { - this.handler.handleTransportError(this, ex); - } - - public final void sendMessage(WebSocketMessage message) throws IOException { - Assert.isTrue(!isClosed(), "Cannot send a message when session is closed"); - Assert.isInstanceOf(TextMessage.class, message, "Expected text message: " + message); - sendMessageInternal(((TextMessage) message).getPayload()); - } - - protected abstract void sendMessageInternal(String message) throws IOException; - - /** - * {@inheritDoc} - *

Perform cleanup and notify the {@link WebSocketHandler}. + * Performs cleanup and notify the {@link WebSocketHandler}. */ @Override public final void close() throws IOException { @@ -263,8 +191,7 @@ public abstract class AbstractSockJsSession implements SockJsSession { } /** - * {@inheritDoc} - *

Perform cleanup and notify the {@link WebSocketHandler}. + * Performs cleanup and notify the {@link WebSocketHandler}. */ @Override public final void close(CloseStatus status) throws IOException { @@ -297,86 +224,28 @@ public abstract class AbstractSockJsSession implements SockJsSession { } } - /** - * Actually close the underlying WebSocket session or in the case of HTTP - * transports complete the underlying request. - */ - protected abstract void disconnect(CloseStatus status) throws IOException; - - /** - * Close due to error arising from SockJS transport handling. - */ - public void tryCloseWithSockJsTransportError(Throwable error, CloseStatus closeStatus) { - if (logger.isDebugEnabled()) { - logger.debug("Closing due to transport error for " + this); - } - try { - delegateError(error); - } - catch (Throwable delegateException) { - // ignore - } - try { - close(closeStatus); - } - catch (Throwable closeException) { - logger.debug("Failure while closing " + this, closeException); - } - } - - /** - * For internal use within a TransportHandler and the (TransportHandler-specific) - * session class. - */ - protected void writeFrame(SockJsFrame frame) throws SockJsTransportFailureException { - if (logger.isTraceEnabled()) { - logger.trace("Preparing to write " + frame); - } - try { - writeFrameInternal(frame); - } - catch (Throwable ex) { - logWriteFrameFailure(ex); - try { - // Force disconnect (so we won't try to send close frame) - disconnect(CloseStatus.SERVER_ERROR); - } - catch (Throwable disconnectFailure) { - // Ignore - } - try { - close(CloseStatus.SERVER_ERROR); - } - catch (Throwable t) { - // Nothing of consequence, already forced disconnect - } - throw new SockJsTransportFailureException("Failed to write " + frame, this.getId(), ex); - } - } - - private void logWriteFrameFailure(Throwable failure) { - - @SuppressWarnings("serial") - NestedCheckedException nestedException = new NestedCheckedException("", failure) {}; - - if ("Broken pipe".equalsIgnoreCase(nestedException.getMostSpecificCause().getMessage()) || - disconnectedClientExceptions.contains(failure.getClass().getSimpleName())) { - - if (disconnectedClientLogger.isTraceEnabled()) { - disconnectedClientLogger.trace("Looks like the client has gone away", failure); - } - else if (disconnectedClientLogger.isDebugEnabled()) { - disconnectedClientLogger.debug("Looks like the client has gone away: " + - nestedException.getMessage() + " (For full stack trace, set the '" + - DISCONNECTED_CLIENT_LOG_CATEGORY + "' log category to TRACE level)"); - } + @Override + public long getTimeSinceLastActive() { + if (isNew()) { + return (System.currentTimeMillis() - this.timeCreated); } else { - logger.debug("Terminating connection after failure to send message to client.", failure); + return (isActive() ? 0 : System.currentTimeMillis() - this.timeLastActive); } } - protected abstract void writeFrameInternal(SockJsFrame frame) throws IOException; + /** + * Should be invoked whenever the session becomes inactive. + */ + protected void updateLastActiveTime() { + this.timeLastActive = System.currentTimeMillis(); + } + + @Override + public void disableHeartbeat() { + this.heartbeatDisabled = true; + cancelHeartbeat(); + } public void sendHeartbeat() throws SockJsTransportFailureException { if (isActive()) { @@ -389,11 +258,13 @@ public abstract class AbstractSockJsSession implements SockJsSession { if (this.heartbeatDisabled) { return; } - Assert.state(this.config.getTaskScheduler() != null, "Expecteded SockJS TaskScheduler."); + + Assert.state(this.config.getTaskScheduler() != null, "Expected SockJS TaskScheduler"); cancelHeartbeat(); if (!isActive()) { return; } + Date time = new Date(System.currentTimeMillis() + this.config.getHeartbeatTime()); this.heartbeatTask = this.config.getTaskScheduler().schedule(new Runnable() { public void run() { @@ -427,12 +298,150 @@ public abstract class AbstractSockJsSession implements SockJsSession { } } + /** + * Polling and Streaming sessions periodically close the current HTTP request and + * wait for the next request to come through. During this "downtime" the session is + * still open but inactive and unable to send messages and therefore has to buffer + * them temporarily. A WebSocket session by contrast is stateful and remain active + * until closed. + */ + public abstract boolean isActive(); + + /** + * Actually close the underlying WebSocket session or in the case of HTTP + * transports complete the underlying request. + */ + protected abstract void disconnect(CloseStatus status) throws IOException; + + + // Frame writing + + /** + * For internal use within a TransportHandler and the (TransportHandler-specific) + * session class. + */ + protected void writeFrame(SockJsFrame frame) throws SockJsTransportFailureException { + if (logger.isTraceEnabled()) { + logger.trace("Preparing to write " + frame); + } + try { + writeFrameInternal(frame); + } + catch (Throwable ex) { + logWriteFrameFailure(ex); + try { + // Force disconnect (so we won't try to send close frame) + disconnect(CloseStatus.SERVER_ERROR); + } + catch (Throwable disconnectFailure) { + // Ignore + } + try { + close(CloseStatus.SERVER_ERROR); + } + catch (Throwable closeFailure) { + // Nothing of consequence, already forced disconnect + } + throw new SockJsTransportFailureException("Failed to write " + frame, this.getId(), ex); + } + } + + private void logWriteFrameFailure(Throwable failure) { + @SuppressWarnings("serial") + NestedCheckedException nestedException = new NestedCheckedException("", failure) {}; + + if ("Broken pipe".equalsIgnoreCase(nestedException.getMostSpecificCause().getMessage()) || + disconnectedClientExceptions.contains(failure.getClass().getSimpleName())) { + + if (disconnectedClientLogger.isTraceEnabled()) { + disconnectedClientLogger.trace("Looks like the client has gone away", failure); + } + else if (disconnectedClientLogger.isDebugEnabled()) { + disconnectedClientLogger.debug("Looks like the client has gone away: " + + nestedException.getMessage() + " (For full stack trace, set the '" + + DISCONNECTED_CLIENT_LOG_CATEGORY + "' log category to TRACE level)"); + } + } + else { + logger.debug("Terminating connection after failure to send message to client.", failure); + } + } + + protected abstract void writeFrameInternal(SockJsFrame frame) throws IOException; + + + // Delegation methods + + public void delegateConnectionEstablished() throws Exception { + this.state = State.OPEN; + this.handler.afterConnectionEstablished(this); + } + + public void delegateMessages(String... messages) throws SockJsMessageDeliveryException { + List undelivered = new ArrayList(Arrays.asList(messages)); + for (String message : messages) { + try { + if (isClosed()) { + throw new SockJsMessageDeliveryException(this.id, undelivered, "Session closed"); + } + else { + this.handler.handleMessage(this, new TextMessage(message)); + undelivered.remove(0); + } + } + catch (Throwable ex) { + throw new SockJsMessageDeliveryException(this.id, undelivered, ex); + } + } + } + + /** + * Invoked when the underlying connection is closed. + */ + public final void delegateConnectionClosed(CloseStatus status) throws Exception { + if (!isClosed()) { + try { + updateLastActiveTime(); + cancelHeartbeat(); + } + finally { + this.state = State.CLOSED; + this.handler.afterConnectionClosed(this, status); + } + } + } + + /** + * Close due to error arising from SockJS transport handling. + */ + public void tryCloseWithSockJsTransportError(Throwable error, CloseStatus closeStatus) { + if (logger.isDebugEnabled()) { + logger.debug("Closing due to transport error for " + this); + } + try { + delegateError(error); + } + catch (Throwable delegateException) { + // ignore + } + try { + close(closeStatus); + } + catch (Throwable closeException) { + logger.debug("Failure while closing " + this, closeException); + } + } + + public void delegateError(Throwable ex) throws Exception { + this.handler.handleTransportError(this, ex); + } + + + // Self description + @Override public String toString() { return getClass().getSimpleName() + "[id=" + getId() + "]"; } - - private enum State { NEW, OPEN, CLOSED } - }