|
|
|
|
@@ -135,9 +135,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
* Since a STOMP message can be received in multiple WebSocket messages,
|
|
|
|
|
* buffering may be required and therefore it is necessary to know the maximum
|
|
|
|
|
* allowed message size.
|
|
|
|
|
*
|
|
|
|
|
* <p>By default this property is set to 64K.
|
|
|
|
|
*
|
|
|
|
|
* @since 4.0.3
|
|
|
|
|
*/
|
|
|
|
|
public void setMessageSizeLimit(int messageSizeLimit) {
|
|
|
|
|
@@ -146,14 +144,12 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Get the configured message buffer size limit in bytes.
|
|
|
|
|
*
|
|
|
|
|
* @since 4.0.3
|
|
|
|
|
*/
|
|
|
|
|
public int getMessageSizeLimit() {
|
|
|
|
|
return this.messageSizeLimit;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Provide a registry with which to register active user session ids.
|
|
|
|
|
* @see org.springframework.messaging.simp.user.UserDestinationMessageHandler
|
|
|
|
|
@@ -179,7 +175,6 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
* Configure a {@link MessageHeaderInitializer} to apply to the headers of all
|
|
|
|
|
* messages created from decoded STOMP frames and other messages sent to the
|
|
|
|
|
* client inbound channel.
|
|
|
|
|
*
|
|
|
|
|
* <p>By default this property is not set.
|
|
|
|
|
*/
|
|
|
|
|
public void setHeaderInitializer(MessageHeaderInitializer headerInitializer) {
|
|
|
|
|
@@ -188,7 +183,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* @return the configured header initializer.
|
|
|
|
|
* Return the configured header initializer.
|
|
|
|
|
*/
|
|
|
|
|
public MessageHeaderInitializer getHeaderInitializer() {
|
|
|
|
|
return this.headerInitializer;
|
|
|
|
|
@@ -259,7 +254,6 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
try {
|
|
|
|
|
StompHeaderAccessor headerAccessor =
|
|
|
|
|
MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
|
|
|
|
|
|
|
|
|
|
Principal user = session.getPrincipal();
|
|
|
|
|
|
|
|
|
|
headerAccessor.setSessionId(session.getId());
|
|
|
|
|
@@ -314,34 +308,37 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
sendErrorMessage(session, ex);
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
Message<byte[]> message = getErrorHandler().handleClientMessageProcessingError(clientMessage, ex);
|
|
|
|
|
if (message == null) {
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
|
|
|
|
|
Assert.notNull(accessor, "Expected STOMP headers.");
|
|
|
|
|
Assert.notNull(accessor, "Expected STOMP headers");
|
|
|
|
|
sendToClient(session, accessor, message.getPayload());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/**
|
|
|
|
|
* Invoked when no
|
|
|
|
|
* {@link #setErrorHandler(StompSubProtocolErrorHandler) errorHandler} is
|
|
|
|
|
* configured to send an ERROR frame to the client.
|
|
|
|
|
* @deprecated as of 4.2 this method is deprecated in favor of
|
|
|
|
|
* {@link #setErrorHandler(StompSubProtocolErrorHandler) configuring} a
|
|
|
|
|
* {@code StompSubProtocolErrorHandler}.
|
|
|
|
|
* {@link #setErrorHandler(StompSubProtocolErrorHandler) errorHandler}
|
|
|
|
|
* is configured to send an ERROR frame to the client.
|
|
|
|
|
* @deprecated as of Spring 4.2, in favor of
|
|
|
|
|
* {@link #setErrorHandler(StompSubProtocolErrorHandler) configuring}
|
|
|
|
|
* a {@code StompSubProtocolErrorHandler}
|
|
|
|
|
*/
|
|
|
|
|
@Deprecated
|
|
|
|
|
protected void sendErrorMessage(WebSocketSession session, Throwable error) {
|
|
|
|
|
StompHeaderAccessor headerAccessor = StompHeaderAccessor.create(StompCommand.ERROR);
|
|
|
|
|
headerAccessor.setMessage(error.getMessage());
|
|
|
|
|
|
|
|
|
|
byte[] bytes = this.stompEncoder.encode(headerAccessor.getMessageHeaders(), EMPTY_PAYLOAD);
|
|
|
|
|
try {
|
|
|
|
|
session.sendMessage(new TextMessage(bytes));
|
|
|
|
|
}
|
|
|
|
|
catch (Throwable ex) {
|
|
|
|
|
// Could be part of normal workflow (e.g. browser tab closed)
|
|
|
|
|
logger.debug("Failed to send STOMP ERROR to client.", ex);
|
|
|
|
|
logger.debug("Failed to send STOMP ERROR to client", ex);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -349,6 +346,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
if (this.immutableMessageInterceptorPresent != null) {
|
|
|
|
|
return this.immutableMessageInterceptorPresent;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (channel instanceof AbstractMessageChannel) {
|
|
|
|
|
for (ChannelInterceptor interceptor : ((AbstractMessageChannel) channel).getInterceptors()) {
|
|
|
|
|
if (interceptor instanceof ImmutableMessageChannelInterceptor) {
|
|
|
|
|
@@ -366,7 +364,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
this.eventPublisher.publishEvent(event);
|
|
|
|
|
}
|
|
|
|
|
catch (Throwable ex) {
|
|
|
|
|
logger.error("Error publishing " + event + ".", ex);
|
|
|
|
|
logger.error("Error publishing " + event, ex);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -380,8 +378,10 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
logger.error("Expected byte[] payload. Ignoring " + message + ".");
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
StompHeaderAccessor stompAccessor = getStompHeaderAccessor(message);
|
|
|
|
|
StompCommand command = stompAccessor.getCommand();
|
|
|
|
|
|
|
|
|
|
if (StompCommand.MESSAGE.equals(command)) {
|
|
|
|
|
if (stompAccessor.getSubscriptionId() == null) {
|
|
|
|
|
logger.warn("No STOMP \"subscription\" header in " + message);
|
|
|
|
|
@@ -414,7 +414,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
if (StompCommand.ERROR.equals(command) && getErrorHandler() != null) {
|
|
|
|
|
Message<byte[]> errorMessage = getErrorHandler().handleErrorMessageToClient((Message<byte[]>) message);
|
|
|
|
|
stompAccessor = MessageHeaderAccessor.getAccessor(errorMessage, StompHeaderAccessor.class);
|
|
|
|
|
Assert.notNull(stompAccessor, "Expected STOMP headers.");
|
|
|
|
|
Assert.notNull(stompAccessor, "Expected STOMP headers");
|
|
|
|
|
payload = errorMessage.getPayload();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -442,7 +442,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
}
|
|
|
|
|
catch (Throwable ex) {
|
|
|
|
|
// Could be part of normal workflow (e.g. browser tab closed)
|
|
|
|
|
logger.debug("Failed to send WebSocket message to client in session " + session.getId() + ".", ex);
|
|
|
|
|
logger.debug("Failed to send WebSocket message to client in session " + session.getId(), ex);
|
|
|
|
|
command = StompCommand.ERROR;
|
|
|
|
|
}
|
|
|
|
|
finally {
|
|
|
|
|
@@ -461,7 +461,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class);
|
|
|
|
|
if (accessor == null) {
|
|
|
|
|
// Shouldn't happen (only broker broadcasts directly to clients)
|
|
|
|
|
throw new IllegalStateException("No header accessor in " + message + ".");
|
|
|
|
|
throw new IllegalStateException("No header accessor in " + message);
|
|
|
|
|
}
|
|
|
|
|
StompHeaderAccessor stompAccessor;
|
|
|
|
|
if (accessor instanceof StompHeaderAccessor) {
|
|
|
|
|
@@ -487,7 +487,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
else {
|
|
|
|
|
// Shouldn't happen (only broker broadcasts directly to clients)
|
|
|
|
|
throw new IllegalStateException(
|
|
|
|
|
"Unexpected header accessor type: " + accessor.getClass() + " in " + message + ".");
|
|
|
|
|
"Unexpected header accessor type: " + accessor.getClass() + " in " + message);
|
|
|
|
|
}
|
|
|
|
|
return stompAccessor;
|
|
|
|
|
}
|
|
|
|
|
@@ -515,7 +515,8 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
long[] heartbeat = (long[]) connectAckHeaders.getHeader(SimpMessageHeaderAccessor.HEART_BEAT_HEADER);
|
|
|
|
|
if (heartbeat != null) {
|
|
|
|
|
connectedHeaders.setHeartbeat(heartbeat[0], heartbeat[1]);
|
|
|
|
|
} else {
|
|
|
|
|
}
|
|
|
|
|
else {
|
|
|
|
|
connectedHeaders.setHeartbeat(0, 0);
|
|
|
|
|
}
|
|
|
|
|
return connectedHeaders;
|
|
|
|
|
@@ -537,6 +538,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
this.userSessionRegistry.registerSessionId(userName, session.getId());
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
long[] heartbeat = accessor.getHeartbeat();
|
|
|
|
|
if (heartbeat[1] > 0) {
|
|
|
|
|
session = WebSocketSessionDecorator.unwrap(session);
|
|
|
|
|
@@ -544,6 +546,7 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
((SockJsSession) session).disableHeartbeat();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return accessor;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -571,11 +574,13 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
@Override
|
|
|
|
|
public void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus, MessageChannel outputChannel) {
|
|
|
|
|
this.decoders.remove(session.getId());
|
|
|
|
|
|
|
|
|
|
Principal principal = session.getPrincipal();
|
|
|
|
|
if (principal != null && this.userSessionRegistry != null) {
|
|
|
|
|
String userName = getSessionRegistryUserName(principal);
|
|
|
|
|
this.userSessionRegistry.unregisterSessionId(userName, session.getId());
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
Message<byte[]> message = createDisconnectMessage(session);
|
|
|
|
|
SimpAttributes simpAttributes = SimpAttributes.fromMessage(message);
|
|
|
|
|
try {
|
|
|
|
|
@@ -608,7 +613,8 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
return "StompSubProtocolHandler" + getSupportedProtocols();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private class Stats {
|
|
|
|
|
|
|
|
|
|
private static class Stats {
|
|
|
|
|
|
|
|
|
|
private final AtomicInteger connect = new AtomicInteger();
|
|
|
|
|
|
|
|
|
|
@@ -616,7 +622,6 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
|
|
|
|
|
private final AtomicInteger disconnect = new AtomicInteger();
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public void incrementConnectCount() {
|
|
|
|
|
this.connect.incrementAndGet();
|
|
|
|
|
}
|
|
|
|
|
@@ -629,7 +634,6 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
|
|
|
|
|
this.disconnect.incrementAndGet();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public String toString() {
|
|
|
|
|
return "processed CONNECT(" + this.connect.get() + ")-CONNECTED(" +
|
|
|
|
|
this.connected.get() + ")-DISCONNECT(" + this.disconnect.get() + ")";
|
|
|
|
|
|