Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

WebSocketSession extends java.io.Closeable, plus reorganization of AbstractSockJsSession's code

Browse Source 
Issue: SPR-12311
This commit is contained in:
Juergen Hoeller
2014-10-07 13:45:07 +02:00
parent da14aeea7a
commit e56559fd4d
2 changed files with 193 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-websocket/src/main/java/org/springframework/web/socket/WebSocketSession.java
View File

@@ -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.
*
* <p>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<String, Object> getAttributes();
@@ -109,23 +108,23 @@ public interface WebSocketSession {
*/
List<WebSocketExtension> 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:
* <pre class="code">
* session.close(CloseStatus.NORMAL);
* </pre>
*/
@Override
void close() throws IOException;
/**

357
spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/session/AbstractSockJsSession.java
View File

@@ -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.
*
* <p>The Servlet API does not provide notifications when a client disconnects;
* see <a href="https://java.net/jira/browse/SERVLET_SPEC-44">SERVLET_SPEC-44</a>.
* Therefore network IO failures may occur simply because a client has gone away,
* and that can fill the logs with unnecessary stack traces.
*
* <p>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<String> 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<String, Object> attributes = new ConcurrentHashMap<String, Object>();
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<String> undelivered = new ArrayList<String>(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}
* <p>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}
* <p>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<String> undelivered = new ArrayList<String>(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 }
}