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

Add STOMP subscribe/unscubscribe ApplicationContext events

Browse Source 
Issue: SPR-11813
This commit is contained in:
Rossen Stoyanchev
2014-06-28 11:07:42 -04:00
parent 9aa53abdf9
commit 1c91a52639
9 changed files with 178 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
spring-websocket/src/main/java/org/springframework/web/socket/messaging/AbstractSubProtocolEvent.java Normal file
View File

@@ -0,0 +1,71 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.web.socket.messaging;
import org.springframework.context.ApplicationEvent;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* A base class for events for a message received from a WebSocket client and
* parsed into a higher level sub-protocol (e.g. STOMP).
*
* @author Rossen Stoyanchev
* @since 4.0.3
*/
@SuppressWarnings("serial")
public abstract class AbstractSubProtocolEvent extends ApplicationEvent {
private final Message<byte[]> message;
/**
* Create a new SessionConnectEvent.
*
* @param source the component that published the event (never {@code null})
* @param message the connect message
*/
protected AbstractSubProtocolEvent(Object source, Message<byte[]> message) {
super(source);
Assert.notNull(message, "'message' must not be null");
this.message = message;
}
/**
* Return the Message associated with the event. Here is an example of
* obtaining information about the session id or any headers in the
* message:
*
* <pre class="code">
* StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
* headers.getSessionId();
* headers.getSessionAttributes();
* headers.getPrincipal();
* </pre>
*
*/
public Message<byte[]> getMessage() {
return this.message;
}
@Override
public String toString() {
return getClass().getSimpleName() + "[" + this.message + "]";
}
}

41
spring-websocket/src/main/java/org/springframework/web/socket/messaging/SessionConnectEvent.java
View File

@@ -17,9 +17,7 @@
package org.springframework.web.socket.messaging;
import org.springframework.context.ApplicationEvent;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* Event raised when a new WebSocket client using a Simple Messaging Protocol
@@ -29,50 +27,15 @@ import org.springframework.util.Assert;
* but rather the client's first attempt to connect within the the sub-protocol,
* for example sending the STOMP CONNECT frame.
*
* <p>The provided {@link #getMessage() message} can be examined to check
* information about the connected user, The session id, and any headers
* sent by the client, for STOMP check the class
* {@link org.springframework.messaging.simp.stomp.StompHeaderAccessor}.
* For example:
*
* <pre class="code">
* StompHeaderAccessor headers = StompHeaderAccessor.wrap(message);
* headers.getSessionId();
* headers.getSessionAttributes();
* headers.getPrincipal();
* </pre>
*
* @author Rossen Stoyanchev
* @since 4.0.3
*/
@SuppressWarnings("serial")
public class SessionConnectEvent extends ApplicationEvent {
private final Message<byte[]> message;
public class SessionConnectEvent extends AbstractSubProtocolEvent {
/**
* Create a new SessionConnectEvent.
*
* @param source the component that published the event (never {@code null})
* @param message the connect message
*/
public SessionConnectEvent(Object source, Message<byte[]> message) {
super(source);
Assert.notNull(message, "'message' must not be null");
this.message = message;
super(source, message);
}
/**
* Return the connect message.
*/
public Message<byte[]> getMessage() {
return this.message;
}
@Override
public String toString() {
return "SessionConnectEvent" + this.message;
}
}

31
spring-websocket/src/main/java/org/springframework/web/socket/messaging/SessionConnectedEvent.java
View File

@@ -17,45 +17,22 @@
package org.springframework.web.socket.messaging;
import org.springframework.context.ApplicationEvent;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* A connected event represents the server response to a client's connect request.
* See {@link org.springframework.web.socket.messaging.SessionConnectEvent}.
* See {@link org.springframework.web.socket.messaging.SessionConnectEvent
* SessionConnectEvent}.
*
* @author Rossen Stoyanchev
* @since 4.0.3
*/
@SuppressWarnings("serial")
public class SessionConnectedEvent extends ApplicationEvent {
private final Message<byte[]> message;
public class SessionConnectedEvent extends AbstractSubProtocolEvent {
/**
* Create a new event.
*
* @param source the component that published the event (never {@code null})
* @param message the connected message
*/
public SessionConnectedEvent(Object source, Message<byte[]> message) {
super(source);
Assert.notNull(message, "'message' must not be null");
this.message = message;
super(source, message);
}
/**
* Return the connected message.
*/
public Message<byte[]> getMessage() {
return this.message;
}
@Override
public String toString() {
return "SessionConnectedEvent" + this.message;
}
}

2
spring-websocket/src/main/java/org/springframework/web/socket/messaging/SessionDisconnectEvent.java
View File

@@ -68,7 +68,7 @@ public class SessionDisconnectEvent extends ApplicationEvent {
@Override
public String toString() {
return "SessionDisconnectEvent[sessionId=" + this.sessionId +
return "SessionDisconnectEvent[sessionId=" + this.sessionId + ", " +
(this.status != null ? this.status.toString() : "closeStatus=null") + "]";
}
}

37
spring-websocket/src/main/java/org/springframework/web/socket/messaging/SessionSubscribeEvent.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.web.socket.messaging;
import org.springframework.messaging.Message;
/**
* Event raised when a new WebSocket client using a Simple Messaging Protocol
* (e.g. STOMP) sends a subscription request.
*
* @author Rossen Stoyanchev
* @since 4.0.3
*/
@SuppressWarnings("serial")
public class SessionSubscribeEvent extends AbstractSubProtocolEvent {
public SessionSubscribeEvent(Object source, Message<byte[]> message) {
super(source, message);
}
}

37
spring-websocket/src/main/java/org/springframework/web/socket/messaging/SessionUnsubscribeEvent.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.web.socket.messaging;
import org.springframework.messaging.Message;
/**
* Event raised when a new WebSocket client using a Simple Messaging Protocol
* (e.g. STOMP) sends a request to remove a subscription.
*
* @author Rossen Stoyanchev
* @since 4.0.3
*/
@SuppressWarnings("serial")
public class SessionUnsubscribeEvent extends AbstractSubProtocolEvent {
public SessionUnsubscribeEvent(Object source, Message<byte[]> message) {
super(source, message);
}
}

12
spring-websocket/src/main/java/org/springframework/web/socket/messaging/StompSubProtocolHandler.java
View File

@@ -219,8 +219,16 @@ public class StompSubProtocolHandler implements SubProtocolHandler, ApplicationE
headerAccessor.setUser(session.getPrincipal());
headerAccessor.setImmutable();
if (this.eventPublisher != null && StompCommand.CONNECT.equals(headerAccessor.getCommand())) {
publishEvent(new SessionConnectEvent(this, message));
if (this.eventPublisher != null) {
if (StompCommand.CONNECT.equals(headerAccessor.getCommand())) {
publishEvent(new SessionConnectEvent(this, message));
}
else if (StompCommand.SUBSCRIBE.equals(headerAccessor.getCommand())) {
publishEvent(new SessionSubscribeEvent(this, message));
}
else if (StompCommand.UNSUBSCRIBE.equals(headerAccessor.getCommand())) {
publishEvent(new SessionUnsubscribeEvent(this, message));
}
}
try {

16
spring-websocket/src/test/java/org/springframework/web/socket/messaging/StompSubProtocolHandlerTests.java
View File

@@ -184,12 +184,24 @@ public class StompSubProtocolHandlerTests {
message = MessageBuilder.createMessage(EMPTY_PAYLOAD, headers.getMessageHeaders());
this.protocolHandler.handleMessageToClient(this.session, message);
headers = StompHeaderAccessor.create(StompCommand.SUBSCRIBE);
message = MessageBuilder.createMessage(EMPTY_PAYLOAD, headers.getMessageHeaders());
textMessage = new TextMessage(new StompEncoder().encode(message));
this.protocolHandler.handleMessageFromClient(this.session, textMessage, this.channel);
headers = StompHeaderAccessor.create(StompCommand.UNSUBSCRIBE);
message = MessageBuilder.createMessage(EMPTY_PAYLOAD, headers.getMessageHeaders());
textMessage = new TextMessage(new StompEncoder().encode(message));
this.protocolHandler.handleMessageFromClient(this.session, textMessage, this.channel);
this.protocolHandler.afterSessionEnded(this.session, CloseStatus.BAD_DATA, this.channel);
assertEquals("Unexpected events " + publisher.events, 3, publisher.events.size());
assertEquals("Unexpected events " + publisher.events, 5, publisher.events.size());
assertEquals(SessionConnectEvent.class, publisher.events.get(0).getClass());
assertEquals(SessionConnectedEvent.class, publisher.events.get(1).getClass());
assertEquals(SessionDisconnectEvent.class, publisher.events.get(2).getClass());
assertEquals(SessionSubscribeEvent.class, publisher.events.get(2).getClass());
assertEquals(SessionUnsubscribeEvent.class, publisher.events.get(3).getClass());
assertEquals(SessionDisconnectEvent.class, publisher.events.get(4).getClass());
}
@Test