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

Add generic parameters to MessageHandler impls

Browse Source
This commit is contained in:
Rossen Stoyanchev
2013-06-18 20:35:15 -04:00
parent f6398e14d0
commit 3f9da6f480
18 changed files with 185 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-websocket/src/main/java/org/springframework/web/messaging/PubSubChannelRegistry.java
View File

@@ -25,7 +25,8 @@ import org.springframework.messaging.SubscribableChannel;
* @author Rossen Stoyanchev
* @since 4.0
*/
public interface PubSubChannelRegistry<M extends Message<?>, H extends MessageHandler<M>> {
@SuppressWarnings("rawtypes")
public interface PubSubChannelRegistry<M extends Message, H extends MessageHandler<M>> {
SubscribableChannel<M, H> getClientInputChannel();

21
spring-websocket/src/main/java/org/springframework/web/messaging/service/AbstractPubSubMessageHandler.java
View File

@@ -37,7 +37,8 @@ import org.springframework.web.messaging.PubSubHeaders;
* @author Rossen Stoyanchev
* @since 4.0
*/
public abstract class AbstractPubSubMessageHandler implements MessageHandler<Message<?>> {
@SuppressWarnings("rawtypes")
public abstract class AbstractPubSubMessageHandler<M extends Message> implements MessageHandler<M> {
protected final Log logger = LogFactory.getLog(getClass());
@@ -67,7 +68,7 @@ public abstract class AbstractPubSubMessageHandler implements MessageHandler<Mes
protected abstract Collection<MessageType> getSupportedMessageTypes();
protected boolean canHandle(Message<?> message, MessageType messageType) {
protected boolean canHandle(M message, MessageType messageType) {
if (!CollectionUtils.isEmpty(getSupportedMessageTypes())) {
if (!getSupportedMessageTypes().contains(messageType)) {
@@ -78,7 +79,7 @@ public abstract class AbstractPubSubMessageHandler implements MessageHandler<Mes
return isDestinationAllowed(message);
}
protected boolean isDestinationAllowed(Message<?> message) {
protected boolean isDestinationAllowed(M message) {
PubSubHeaders headers = PubSubHeaders.fromMessageHeaders(message.getHeaders());
String destination = headers.getDestination();
@@ -114,7 +115,7 @@ public abstract class AbstractPubSubMessageHandler implements MessageHandler<Mes
}
@Override
public final void handleMessage(Message<?> message) throws MessagingException {
public final void handleMessage(M message) throws MessagingException {
PubSubHeaders headers = PubSubHeaders.fromMessageHeaders(message.getHeaders());
MessageType messageType = headers.getMessageType();
@@ -143,22 +144,22 @@ public abstract class AbstractPubSubMessageHandler implements MessageHandler<Mes
}
}
protected void handleConnect(Message<?> message) {
protected void handleConnect(M message) {
}
protected void handlePublish(Message<?> message) {
protected void handlePublish(M message) {
}
protected void handleSubscribe(Message<?> message) {
protected void handleSubscribe(M message) {
}
protected void handleUnsubscribe(Message<?> message) {
protected void handleUnsubscribe(M message) {
}
protected void handleDisconnect(Message<?> message) {
protected void handleDisconnect(M message) {
}
protected void handleOther(Message<?> message) {
protected void handleOther(M message) {
}
}

27
spring-websocket/src/main/java/org/springframework/web/messaging/service/ReactorPubSubMessageHandler.java
View File

@@ -44,9 +44,10 @@ import reactor.fn.selector.ObjectSelector;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
@SuppressWarnings("rawtypes")
public class ReactorPubSubMessageHandler<M extends Message> extends AbstractPubSubMessageHandler<M> {
private MessageChannel<Message<?>> clientChannel;
private MessageChannel<M> clientChannel;
private final Reactor reactor;
@@ -55,7 +56,7 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
private Map<String, List<Registration<?>>> subscriptionsBySession = new ConcurrentHashMap<String, List<Registration<?>>>();
public ReactorPubSubMessageHandler(PubSubChannelRegistry registry, Reactor reactor) {
public ReactorPubSubMessageHandler(PubSubChannelRegistry<M, ?> registry, Reactor reactor) {
Assert.notNull(reactor, "reactor is required");
this.clientChannel = registry.getClientOutputChannel();
this.reactor = reactor;
@@ -72,7 +73,7 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
}
@Override
public void handleSubscribe(Message<?> message) {
public void handleSubscribe(M message) {
if (logger.isDebugEnabled()) {
logger.debug("Subscribe " + message);
@@ -99,7 +100,7 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
}
@Override
public void handlePublish(Message<?> message) {
public void handlePublish(M message) {
if (logger.isDebugEnabled()) {
logger.debug("Message received: " + message);
@@ -109,9 +110,10 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
// Convert to byte[] payload before the fan-out
PubSubHeaders headers = PubSubHeaders.fromMessageHeaders(message.getHeaders());
byte[] payload = payloadConverter.convertToPayload(message.getPayload(), headers.getContentType());
message = MessageBuilder.fromPayloadAndHeaders(payload, message.getHeaders()).build();
@SuppressWarnings("unchecked")
M m = (M) MessageBuilder.fromPayloadAndHeaders(payload, message.getHeaders()).build();
this.reactor.notify(getPublishKey(headers.getDestination()), Event.wrap(message));
this.reactor.notify(getPublishKey(headers.getDestination()), Event.wrap(m));
}
catch (Exception ex) {
logger.error("Failed to publish " + message, ex);
@@ -119,17 +121,11 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
}
@Override
public void handleDisconnect(Message<?> message) {
public void handleDisconnect(M message) {
PubSubHeaders headers = PubSubHeaders.fromMessageHeaders(message.getHeaders());
removeSubscriptions(headers.getSessionId());
}
/* @Override
public void handleClientConnectionClosed(String sessionId) {
removeSubscriptions(sessionId);
}
*/
private void removeSubscriptions(String sessionId) {
List<Registration<?>> registrations = this.subscriptionsBySession.remove(sessionId);
if (logger.isTraceEnabled()) {
@@ -158,7 +154,8 @@ public class ReactorPubSubMessageHandler extends AbstractPubSubMessageHandler {
PubSubHeaders clientHeaders = PubSubHeaders.fromMessageHeaders(sentMessage.getHeaders());
clientHeaders.setSubscriptionId(this.subscriptionId);
Message<?> clientMessage = MessageBuilder.fromPayloadAndHeaders(sentMessage.getPayload(),
@SuppressWarnings("unchecked")
M clientMessage = (M) MessageBuilder.fromPayloadAndHeaders(sentMessage.getPayload(),
clientHeaders.toMessageHeaders()).build();
clientChannel.send(clientMessage);

27
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/AnnotationPubSubMessageHandler.java
View File

@@ -52,10 +52,11 @@ import org.springframework.web.method.HandlerMethodSelector;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class AnnotationPubSubMessageHandler extends AbstractPubSubMessageHandler
@SuppressWarnings("rawtypes")
public class AnnotationPubSubMessageHandler<M extends Message> extends AbstractPubSubMessageHandler<M>
implements ApplicationContextAware, InitializingBean {
private PubSubChannelRegistry registry;
private PubSubChannelRegistry<M, ?> registry;
private List<MessageConverter> messageConverters;
@@ -67,12 +68,12 @@ public class AnnotationPubSubMessageHandler extends AbstractPubSubMessageHandler
private Map<MappingInfo, HandlerMethod> unsubscribeMethods = new HashMap<MappingInfo, HandlerMethod>();
private ArgumentResolverComposite argumentResolvers = new ArgumentResolverComposite();
private ArgumentResolverComposite<M> argumentResolvers = new ArgumentResolverComposite<M>();
private ReturnValueHandlerComposite returnValueHandlers = new ReturnValueHandlerComposite();
private ReturnValueHandlerComposite<M> returnValueHandlers = new ReturnValueHandlerComposite<M>();
public AnnotationPubSubMessageHandler(PubSubChannelRegistry registry) {
public AnnotationPubSubMessageHandler(PubSubChannelRegistry<M, ?> registry) {
Assert.notNull(registry, "registry is required");
this.registry = registry;
}
@@ -96,10 +97,10 @@ public class AnnotationPubSubMessageHandler extends AbstractPubSubMessageHandler
initHandlerMethods();
this.argumentResolvers.addResolver(new MessageChannelArgumentResolver(this.registry.getMessageBrokerChannel()));
this.argumentResolvers.addResolver(new MessageBodyArgumentResolver(this.messageConverters));
this.argumentResolvers.addResolver(new MessageChannelArgumentResolver<M>(this.registry.getMessageBrokerChannel()));
this.argumentResolvers.addResolver(new MessageBodyArgumentResolver<M>(this.messageConverters));
this.returnValueHandlers.addHandler(new MessageReturnValueHandler(this.registry.getClientOutputChannel()));
this.returnValueHandlers.addHandler(new MessageReturnValueHandler<M>(this.registry.getClientOutputChannel()));
}
protected void initHandlerMethods() {
@@ -165,21 +166,21 @@ public class AnnotationPubSubMessageHandler extends AbstractPubSubMessageHandler
}
@Override
public void handlePublish(Message<?> message) {
public void handlePublish(M message) {
handleMessageInternal(message, this.messageMethods);
}
@Override
public void handleSubscribe(Message<?> message) {
public void handleSubscribe(M message) {
handleMessageInternal(message, this.subscribeMethods);
}
@Override
public void handleUnsubscribe(Message<?> message) {
public void handleUnsubscribe(M message) {
handleMessageInternal(message, this.unsubscribeMethods);
}
private void handleMessageInternal(final Message<?> message, Map<MappingInfo, HandlerMethod> handlerMethods) {
private void handleMessageInternal(final M message, Map<MappingInfo, HandlerMethod> handlerMethods) {
PubSubHeaders headers = PubSubHeaders.fromMessageHeaders(message.getHeaders());
String destination = headers.getDestination();
@@ -192,7 +193,7 @@ public class AnnotationPubSubMessageHandler extends AbstractPubSubMessageHandler
HandlerMethod handlerMethod = match.createWithResolvedBean();
// TODO:
InvocableMessageHandlerMethod invocableHandlerMethod = new InvocableMessageHandlerMethod(handlerMethod);
InvocableMessageHandlerMethod<M> invocableHandlerMethod = new InvocableMessageHandlerMethod<M>(handlerMethod);
invocableHandlerMethod.setMessageMethodArgumentResolvers(this.argumentResolvers);
try {

5
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/ArgumentResolver.java
View File

@@ -27,7 +27,8 @@ import org.springframework.messaging.Message;
* @author Rossen Stoyanchev
* @since 4.0
*/
public interface ArgumentResolver {
@SuppressWarnings("rawtypes")
public interface ArgumentResolver<M extends Message> {
/**
* Whether the given {@linkplain MethodParameter method parameter} is
@@ -53,6 +54,6 @@ public interface ArgumentResolver {
*
* @throws Exception in case of errors with the preparation of argument values
*/
Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception;
Object resolveArgument(MethodParameter parameter, M message) throws Exception;
}

28
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/ArgumentResolverComposite.java
View File

@@ -36,21 +36,21 @@ import org.springframework.util.Assert;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class ArgumentResolverComposite implements ArgumentResolver {
@SuppressWarnings("rawtypes")
public class ArgumentResolverComposite<M extends Message> implements ArgumentResolver<M> {
protected final Log logger = LogFactory.getLog(getClass());
private final List<ArgumentResolver> argumentResolvers =
new LinkedList<ArgumentResolver>();
private final List<ArgumentResolver<M>> argumentResolvers = new LinkedList<ArgumentResolver<M>>();
private final Map<MethodParameter, ArgumentResolver> argumentResolverCache =
new ConcurrentHashMap<MethodParameter, ArgumentResolver>(256);
private final Map<MethodParameter, ArgumentResolver<M>> argumentResolverCache =
new ConcurrentHashMap<MethodParameter, ArgumentResolver<M>>(256);
/**
* Return a read-only list with the contained resolvers, or an empty list.
*/
public List<ArgumentResolver> getResolvers() {
public List<ArgumentResolver<M>> getResolvers() {
return Collections.unmodifiableList(this.argumentResolvers);
}
@@ -68,9 +68,9 @@ public class ArgumentResolverComposite implements ArgumentResolver {
* @exception IllegalStateException if no suitable {@link ArgumentResolver} is found.
*/
@Override
public Object resolveArgument(MethodParameter parameter, Message message) throws Exception {
public Object resolveArgument(MethodParameter parameter, M message) throws Exception {
ArgumentResolver resolver = getArgumentResolver(parameter);
ArgumentResolver<M> resolver = getArgumentResolver(parameter);
Assert.notNull(resolver, "Unknown parameter type [" + parameter.getParameterType().getName() + "]");
return resolver.resolveArgument(parameter, message);
}
@@ -78,10 +78,10 @@ public class ArgumentResolverComposite implements ArgumentResolver {
/**
* Find a registered {@link ArgumentResolver} that supports the given method parameter.
*/
private ArgumentResolver getArgumentResolver(MethodParameter parameter) {
ArgumentResolver result = this.argumentResolverCache.get(parameter);
private ArgumentResolver<M> getArgumentResolver(MethodParameter parameter) {
ArgumentResolver<M> result = this.argumentResolverCache.get(parameter);
if (result == null) {
for (ArgumentResolver resolver : this.argumentResolvers) {
for (ArgumentResolver<M> resolver : this.argumentResolvers) {
if (resolver.supportsParameter(parameter)) {
result = resolver;
this.argumentResolverCache.put(parameter, result);
@@ -95,7 +95,7 @@ public class ArgumentResolverComposite implements ArgumentResolver {
/**
* Add the given {@link ArgumentResolver}.
*/
public ArgumentResolverComposite addResolver(ArgumentResolver argumentResolver) {
public ArgumentResolverComposite<M> addResolver(ArgumentResolver<M> argumentResolver) {
this.argumentResolvers.add(argumentResolver);
return this;
}
@@ -103,9 +103,9 @@ public class ArgumentResolverComposite implements ArgumentResolver {
/**
* Add the given {@link ArgumentResolver}s.
*/
public ArgumentResolverComposite addResolvers(List<? extends ArgumentResolver> argumentResolvers) {
public ArgumentResolverComposite<M> addResolvers(List<? extends ArgumentResolver<M>> argumentResolvers) {
if (argumentResolvers != null) {
for (ArgumentResolver resolver : argumentResolvers) {
for (ArgumentResolver<M> resolver : argumentResolvers) {
this.argumentResolvers.add(resolver);
}
}

11
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/InvocableMessageHandlerMethod.java
View File

@@ -43,9 +43,10 @@ import org.springframework.web.method.HandlerMethod;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class InvocableMessageHandlerMethod extends HandlerMethod {
@SuppressWarnings("rawtypes")
public class InvocableMessageHandlerMethod<M extends Message> extends HandlerMethod {
private ArgumentResolverComposite argumentResolvers = new ArgumentResolverComposite();
private ArgumentResolverComposite<M> argumentResolvers = new ArgumentResolverComposite<M>();
private ParameterNameDiscoverer parameterNameDiscoverer = new LocalVariableTableParameterNameDiscoverer();
@@ -75,7 +76,7 @@ public class InvocableMessageHandlerMethod extends HandlerMethod {
* Set {@link ArgumentResolver}s to use to use for resolving method
* argument values.
*/
public void setMessageMethodArgumentResolvers(ArgumentResolverComposite argumentResolvers) {
public void setMessageMethodArgumentResolvers(ArgumentResolverComposite<M> argumentResolvers) {
this.argumentResolvers = argumentResolvers;
}
@@ -97,7 +98,7 @@ public class InvocableMessageHandlerMethod extends HandlerMethod {
* @exception Exception raised if no suitable argument resolver can be found, or the
* method raised an exception
*/
public final Object invoke(Message<?> message) throws Exception {
public final Object invoke(M message) throws Exception {
Object[] args = getMethodArgumentValues(message);
@@ -120,7 +121,7 @@ public class InvocableMessageHandlerMethod extends HandlerMethod {
/**
* Get the method argument values for the current request.
*/
private Object[] getMethodArgumentValues(Message<?> message) throws Exception {
private Object[] getMethodArgumentValues(M message) throws Exception {
MethodParameter[] parameters = getMethodParameters();
Object[] args = new Object[parameters.length];

5
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageBodyArgumentResolver.java
View File

@@ -32,7 +32,8 @@ import org.springframework.web.messaging.converter.MessageConverter;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class MessageBodyArgumentResolver implements ArgumentResolver {
@SuppressWarnings("rawtypes")
public class MessageBodyArgumentResolver<M extends Message> implements ArgumentResolver<M> {
private final MessageConverter converter;
@@ -47,7 +48,7 @@ public class MessageBodyArgumentResolver implements ArgumentResolver {
}
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
public Object resolveArgument(MethodParameter parameter, M message) throws Exception {
Object arg = null;

11
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageChannelArgumentResolver.java
View File

@@ -28,12 +28,13 @@ import org.springframework.web.messaging.support.SessionMessageChannel;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class MessageChannelArgumentResolver implements ArgumentResolver {
@SuppressWarnings("rawtypes")
public class MessageChannelArgumentResolver<M extends Message> implements ArgumentResolver<M> {
private MessageChannel<Message<?>> messageBrokerChannel;
private MessageChannel<M> messageBrokerChannel;
public MessageChannelArgumentResolver(MessageChannel<Message<?>> messageBrokerChannel) {
public MessageChannelArgumentResolver(MessageChannel<M> messageBrokerChannel) {
Assert.notNull(messageBrokerChannel, "messageBrokerChannel is required");
this.messageBrokerChannel = messageBrokerChannel;
}
@@ -44,10 +45,10 @@ public class MessageChannelArgumentResolver implements ArgumentResolver {
}
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
public Object resolveArgument(MethodParameter parameter, M message) throws Exception {
Assert.notNull(this.messageBrokerChannel, "messageBrokerChannel is required");
final String sessionId = PubSubHeaders.fromMessageHeaders(message.getHeaders()).getSessionId();
return new SessionMessageChannel(this.messageBrokerChannel, sessionId);
return new SessionMessageChannel<M>(this.messageBrokerChannel, sessionId);
}
}

22
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/MessageReturnValueHandler.java
View File

@@ -28,12 +28,13 @@ import org.springframework.web.messaging.PubSubHeaders;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class MessageReturnValueHandler implements ReturnValueHandler {
@SuppressWarnings("rawtypes")
public class MessageReturnValueHandler<M extends Message> implements ReturnValueHandler<M> {
private MessageChannel<Message<?>> clientChannel;
private MessageChannel<M> clientChannel;
public MessageReturnValueHandler(MessageChannel<Message<?>> clientChannel) {
public MessageReturnValueHandler(MessageChannel<M> clientChannel) {
Assert.notNull(clientChannel, "clientChannel is required");
this.clientChannel = clientChannel;
}
@@ -55,14 +56,14 @@ public class MessageReturnValueHandler implements ReturnValueHandler {
// return Message.class.isAssignableFrom(paramType);
}
@SuppressWarnings("unchecked")
@Override
public void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> message)
public void handleReturnValue(Object returnValue, MethodParameter returnType, M message)
throws Exception {
Assert.notNull(this.clientChannel, "No clientChannel to send messages to");
Message<?> returnMessage = (Message<?>) returnValue;
@SuppressWarnings("unchecked")
M returnMessage = (M) returnValue;
if (returnMessage == null) {
return;
}
@@ -72,7 +73,7 @@ public class MessageReturnValueHandler implements ReturnValueHandler {
this.clientChannel.send(returnMessage);
}
protected Message<?> updateReturnMessage(Message<?> returnMessage, Message<?> message) {
protected M updateReturnMessage(M returnMessage, M message) {
PubSubHeaders headers = PubSubHeaders.fromMessageHeaders(message.getHeaders());
String sessionId = headers.getSessionId();
@@ -89,7 +90,12 @@ public class MessageReturnValueHandler implements ReturnValueHandler {
}
Object payload = returnMessage.getPayload();
return MessageBuilder.fromPayloadAndHeaders(payload, returnHeaders.toMessageHeaders()).build();
return createMessage(returnHeaders, payload);
}
@SuppressWarnings("unchecked")
private M createMessage(PubSubHeaders returnHeaders, Object payload) {
return (M) MessageBuilder.fromPayloadAndHeaders(payload, returnHeaders.toMessageHeaders()).build();
}
}

5
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/ReturnValueHandler.java
View File

@@ -27,7 +27,8 @@ import org.springframework.messaging.Message;
* @author Rossen Stoyanchev
* @since 4.0
*/
public interface ReturnValueHandler {
@SuppressWarnings("rawtypes")
public interface ReturnValueHandler<M extends Message> {
/**
* Whether the given {@linkplain MethodParameter method return type} is
@@ -50,6 +51,6 @@ public interface ReturnValueHandler {
* @param message the message that caused this method to be called
* @throws Exception if the return value handling results in an error
*/
void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> message) throws Exception;
void handleReturnValue(Object returnValue, MethodParameter returnType, M message) throws Exception;
}

20
spring-websocket/src/main/java/org/springframework/web/messaging/service/method/ReturnValueHandlerComposite.java
View File

@@ -28,16 +28,16 @@ import org.springframework.util.Assert;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class ReturnValueHandlerComposite implements ReturnValueHandler {
@SuppressWarnings("rawtypes")
public class ReturnValueHandlerComposite<M extends Message> implements ReturnValueHandler<M> {
private final List<ReturnValueHandler> returnValueHandlers =
new ArrayList<ReturnValueHandler>();
private final List<ReturnValueHandler<M>> returnValueHandlers = new ArrayList<ReturnValueHandler<M>>();
/**
* Add the given {@link ReturnValueHandler}.
*/
public ReturnValueHandlerComposite addHandler(ReturnValueHandler returnValuehandler) {
public ReturnValueHandlerComposite<M> addHandler(ReturnValueHandler<M> returnValuehandler) {
this.returnValueHandlers.add(returnValuehandler);
return this;
}
@@ -45,9 +45,9 @@ public class ReturnValueHandlerComposite implements ReturnValueHandler {
/**
* Add the given {@link ReturnValueHandler}s.
*/
public ReturnValueHandlerComposite addHandlers(List<? extends ReturnValueHandler> handlers) {
public ReturnValueHandlerComposite<M> addHandlers(List<? extends ReturnValueHandler<M>> handlers) {
if (handlers != null) {
for (ReturnValueHandler handler : handlers) {
for (ReturnValueHandler<M> handler : handlers) {
this.returnValueHandlers.add(handler);
}
}
@@ -59,8 +59,8 @@ public class ReturnValueHandlerComposite implements ReturnValueHandler {
return getReturnValueHandler(returnType) != null;
}
private ReturnValueHandler getReturnValueHandler(MethodParameter returnType) {
for (ReturnValueHandler handler : this.returnValueHandlers) {
private ReturnValueHandler<M> getReturnValueHandler(MethodParameter returnType) {
for (ReturnValueHandler<M> handler : this.returnValueHandlers) {
if (handler.supportsReturnType(returnType)) {
return handler;
}
@@ -69,10 +69,10 @@ public class ReturnValueHandlerComposite implements ReturnValueHandler {
}
@Override
public void handleReturnValue(Object returnValue, MethodParameter returnType, Message<?> message)
public void handleReturnValue(Object returnValue, MethodParameter returnType, M message)
throws Exception {
ReturnValueHandler handler = getReturnValueHandler(returnType);
ReturnValueHandler<M> handler = getReturnValueHandler(returnType);
Assert.notNull(handler, "Unknown return value type [" + returnType.getParameterType().getName() + "]");
handler.handleReturnValue(returnValue, returnType, message);
}

27
spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompMessageConverter.java
View File

@@ -37,7 +37,8 @@ import org.springframework.web.messaging.stomp.StompHeaders;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class StompMessageConverter {
@SuppressWarnings("rawtypes")
public class StompMessageConverter<M extends Message> {
private static final Charset STOMP_CHARSET = Charset.forName("UTF-8");
@@ -50,7 +51,7 @@ public class StompMessageConverter {
/**
* @param stompContent a complete STOMP message (without the trailing 0x00) as byte[] or String.
*/
public Message<byte[]> toMessage(Object stompContent, String sessionId) {
public M toMessage(Object stompContent, String sessionId) {
byte[] byteContent = null;
if (stompContent instanceof String) {
@@ -101,7 +102,12 @@ public class StompMessageConverter {
byte[] payload = new byte[totalLength - payloadIndex];
System.arraycopy(byteContent, payloadIndex, payload, 0, totalLength - payloadIndex);
return MessageBuilder.fromPayloadAndHeaders(payload, stompHeaders.toMessageHeaders()).build();
return createMessage(stompHeaders, payload);
}
@SuppressWarnings("unchecked")
private M createMessage(StompHeaders stompHeaders, byte[] payload) {
return (M) MessageBuilder.fromPayloadAndHeaders(payload, stompHeaders.toMessageHeaders()).build();
}
private int findIndexOfPayload(byte[] bytes) {
@@ -131,10 +137,21 @@ public class StompMessageConverter {
return index;
}
public byte[] fromMessage(Message<byte[]> message) {
public byte[] fromMessage(M message) {
byte[] payload;
if (message.getPayload() instanceof byte[]) {
payload = (byte[]) message.getPayload();
}
else {
throw new IllegalArgumentException(
"stompContent is not byte[]: " + message.getPayload().getClass());
}
ByteArrayOutputStream out = new ByteArrayOutputStream();
MessageHeaders messageHeaders = message.getHeaders();
StompHeaders stompHeaders = StompHeaders.fromMessageHeaders(messageHeaders);
try {
out.write(stompHeaders.getStompCommand().toString().getBytes("UTF-8"));
out.write(LF);
@@ -150,7 +167,7 @@ public class StompMessageConverter {
}
}
out.write(LF);
out.write(message.getPayload());
out.write(payload);
out.write(0);
return out.toByteArray();
}

47
spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompRelayPubSubMessageHandler.java
View File

@@ -55,11 +55,12 @@ import reactor.tcp.netty.NettyTcpClient;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler {
@SuppressWarnings("rawtypes")
public class StompRelayPubSubMessageHandler<M extends Message> extends AbstractPubSubMessageHandler<M> {
private MessageChannel<Message<?>> clientChannel;
private MessageChannel<M> clientChannel;
private final StompMessageConverter stompMessageConverter = new StompMessageConverter();
private final StompMessageConverter<M> stompMessageConverter = new StompMessageConverter<M>();
private MessageConverter payloadConverter;
@@ -72,7 +73,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
* @param clientChannel a channel for sending messages from the remote message broker
* back to clients
*/
public StompRelayPubSubMessageHandler(PubSubChannelRegistry registry) {
public StompRelayPubSubMessageHandler(PubSubChannelRegistry<M, ?> registry) {
Assert.notNull(registry, "registry is required");
this.clientChannel = registry.getClientOutputChannel();
@@ -96,7 +97,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
}
@Override
public void handleConnect(Message<?> message) {
public void handleConnect(M message) {
StompHeaders stompHeaders = StompHeaders.fromMessageHeaders(message.getHeaders());
String sessionId = stompHeaders.getSessionId();
if (sessionId == null) {
@@ -108,22 +109,22 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
}
@Override
public void handlePublish(Message<?> message) {
public void handlePublish(M message) {
forwardMessage(message, StompCommand.SEND);
}
@Override
public void handleSubscribe(Message<?> message) {
public void handleSubscribe(M message) {
forwardMessage(message, StompCommand.SUBSCRIBE);
}
@Override
public void handleUnsubscribe(Message<?> message) {
public void handleUnsubscribe(M message) {
forwardMessage(message, StompCommand.UNSUBSCRIBE);
}
@Override
public void handleDisconnect(Message<?> message) {
public void handleDisconnect(M message) {
StompHeaders stompHeaders = StompHeaders.fromMessageHeaders(message.getHeaders());
if (stompHeaders.getStompCommand() != null) {
forwardMessage(message, StompCommand.DISCONNECT);
@@ -136,13 +137,13 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
}
@Override
public void handleOther(Message<?> message) {
public void handleOther(M message) {
StompCommand command = (StompCommand) message.getHeaders().get(PubSubHeaders.PROTOCOL_MESSAGE_TYPE);
Assert.notNull(command, "Expected STOMP command: " + message.getHeaders());
forwardMessage(message, command);
}
private void forwardMessage(Message<?> message, StompCommand command) {
private void forwardMessage(M message, StompCommand command) {
StompHeaders headers = StompHeaders.fromMessageHeaders(message.getHeaders());
headers.setStompCommandIfNotSet(command);
@@ -172,10 +173,10 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
private final AtomicBoolean isConnected = new AtomicBoolean(false);
private final BlockingQueue<Message<?>> messageQueue = new LinkedBlockingQueue<Message<?>>(50);
private final BlockingQueue<M> messageQueue = new LinkedBlockingQueue<M>(50);
public RelaySession(final Message<?> message, final StompHeaders stompHeaders) {
public RelaySession(final M message, final StompHeaders stompHeaders) {
Assert.notNull(message, "message is required");
Assert.notNull(stompHeaders, "stompHeaders is required");
@@ -216,7 +217,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
return;
}
Message<byte[]> message = stompMessageConverter.toMessage(stompFrame, this.sessionId);
M message = stompMessageConverter.toMessage(stompFrame, this.sessionId);
if (logger.isTraceEnabled()) {
logger.trace("Reading message " + message);
}
@@ -240,19 +241,20 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
StompHeaders stompHeaders = StompHeaders.create(StompCommand.ERROR);
stompHeaders.setSessionId(sessionId);
stompHeaders.setMessage(errorText);
Message<byte[]> errorMessage = MessageBuilder.fromPayloadAndHeaders(
new byte[0], stompHeaders.toMessageHeaders()).build();
@SuppressWarnings("unchecked")
M errorMessage = (M) MessageBuilder.fromPayloadAndHeaders(new byte[0], stompHeaders.toMessageHeaders()).build();
clientChannel.send(errorMessage);
}
public void forward(Message<?> message, StompHeaders headers) {
public void forward(M message, StompHeaders headers) {
if (!this.isConnected.get()) {
message = MessageBuilder.fromPayloadAndHeaders(message.getPayload(), headers.toMessageHeaders()).build();
@SuppressWarnings("unchecked")
M m = (M) MessageBuilder.fromPayloadAndHeaders(message.getPayload(), headers.toMessageHeaders()).build();
if (logger.isTraceEnabled()) {
logger.trace("Adding to queue message " + message + ", queue size=" + this.messageQueue.size());
logger.trace("Adding to queue message " + m + ", queue size=" + this.messageQueue.size());
}
this.messageQueue.add(message);
this.messageQueue.add(m);
return;
}
@@ -268,7 +270,7 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
}
private void flushMessages(TcpConnection<String, String> connection) {
List<Message<?>> messages = new ArrayList<Message<?>>();
List<M> messages = new ArrayList<M>();
this.messageQueue.drainTo(messages);
for (Message<?> message : messages) {
StompHeaders headers = StompHeaders.fromMessageHeaders(message.getHeaders());
@@ -284,7 +286,8 @@ public class StompRelayPubSubMessageHandler extends AbstractPubSubMessageHandler
MediaType contentType = headers.getContentType();
byte[] payload = payloadConverter.convertToPayload(message.getPayload(), contentType);
Message<byte[]> byteMessage = MessageBuilder.fromPayloadAndHeaders(payload, headers.toMessageHeaders()).build();
@SuppressWarnings("unchecked")
M byteMessage = (M) MessageBuilder.fromPayloadAndHeaders(payload, headers.toMessageHeaders()).build();
if (logger.isTraceEnabled()) {
logger.trace("Forwarding message " + byteMessage);

42
spring-websocket/src/main/java/org/springframework/web/messaging/stomp/support/StompWebSocketHandler.java
View File

@@ -49,22 +49,24 @@ import reactor.util.Assert;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implements MessageHandler<Message<?>> {
@SuppressWarnings("rawtypes")
public class StompWebSocketHandler<M extends Message> extends TextWebSocketHandlerAdapter
implements MessageHandler<M> {
private static final byte[] EMPTY_PAYLOAD = new byte[0];
private static Log logger = LogFactory.getLog(StompWebSocketHandler.class);
private MessageChannel outputChannel;
private MessageChannel<M> outputChannel;
private final StompMessageConverter stompMessageConverter = new StompMessageConverter();
private final StompMessageConverter<M> stompMessageConverter = new StompMessageConverter<M>();
private final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<String, WebSocketSession>();
private MessageConverter payloadConverter = new CompositeMessageConverter(null);
public StompWebSocketHandler(PubSubChannelRegistry registry) {
public StompWebSocketHandler(PubSubChannelRegistry<M, ?> registry) {
Assert.notNull(registry, "registry is required");
this.outputChannel = registry.getClientInputChannel();
}
@@ -73,7 +75,7 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
this.payloadConverter = new CompositeMessageConverter(converters);
}
public StompMessageConverter getStompMessageConverter() {
public StompMessageConverter<M> getStompMessageConverter() {
return this.stompMessageConverter;
}
@@ -91,12 +93,11 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
/**
* Handle incoming WebSocket messages from clients.
*/
@SuppressWarnings("unchecked")
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage textMessage) {
try {
String payload = textMessage.getPayload();
Message<byte[]> message = this.stompMessageConverter.toMessage(payload, session.getId());
M message = this.stompMessageConverter.toMessage(payload, session.getId());
// TODO: validate size limits
// http://stomp.github.io/stomp-specification-1.2.html#Size_Limits
@@ -139,7 +140,7 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
}
}
protected void handleConnect(final WebSocketSession session, Message<byte[]> message) throws IOException {
protected void handleConnect(final WebSocketSession session, M message) throws IOException {
StompHeaders connectHeaders = StompHeaders.fromMessageHeaders(message.getHeaders());
StompHeaders connectedHeaders = StompHeaders.create(StompCommand.CONNECTED);
@@ -161,25 +162,26 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
// TODO: security
Message<byte[]> connectedMessage = MessageBuilder.fromPayloadAndHeaders(EMPTY_PAYLOAD,
@SuppressWarnings("unchecked")
M connectedMessage = (M) MessageBuilder.fromPayloadAndHeaders(EMPTY_PAYLOAD,
connectedHeaders.toMessageHeaders()).build();
byte[] bytes = getStompMessageConverter().fromMessage(connectedMessage);
session.sendMessage(new TextMessage(new String(bytes, Charset.forName("UTF-8"))));
}
protected void handlePublish(Message<byte[]> stompMessage) {
protected void handlePublish(M stompMessage) {
}
protected void handleSubscribe(Message<byte[]> message) {
protected void handleSubscribe(M message) {
// TODO: need a way to communicate back if subscription was successfully created or
// not in which case an ERROR should be sent back and close the connection
// http://stomp.github.io/stomp-specification-1.2.html#SUBSCRIBE
}
protected void handleUnsubscribe(Message<byte[]> message) {
protected void handleUnsubscribe(M message) {
}
protected void handleDisconnect(Message<byte[]> stompMessage) {
protected void handleDisconnect(M stompMessage) {
}
protected void sendErrorMessage(WebSocketSession session, Throwable error) {
@@ -187,8 +189,8 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
StompHeaders headers = StompHeaders.create(StompCommand.ERROR);
headers.setMessage(error.getMessage());
Message<byte[]> message = MessageBuilder.fromPayloadAndHeaders(EMPTY_PAYLOAD,
headers.toMessageHeaders()).build();
@SuppressWarnings("unchecked")
M message = (M) MessageBuilder.fromPayloadAndHeaders(EMPTY_PAYLOAD, headers.toMessageHeaders()).build();
byte[] bytes = this.stompMessageConverter.fromMessage(message);
try {
@@ -199,13 +201,13 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
}
}
@SuppressWarnings("unchecked")
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
this.sessions.remove(session.getId());
PubSubHeaders headers = PubSubHeaders.create(MessageType.DISCONNECT);
headers.setSessionId(session.getId());
Message<?> message = MessageBuilder.fromPayloadAndHeaders(new byte[0], headers.toMessageHeaders()).build();
@SuppressWarnings("unchecked")
M message = (M) MessageBuilder.fromPayloadAndHeaders(new byte[0], headers.toMessageHeaders()).build();
this.outputChannel.send(message);
}
@@ -213,7 +215,7 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
* Handle STOMP messages going back out to WebSocket clients.
*/
@Override
public void handleMessage(Message<?> message) {
public void handleMessage(M message) {
StompHeaders headers = StompHeaders.fromMessageHeaders(message.getHeaders());
headers.setStompCommandIfNotSet(StompCommand.MESSAGE);
@@ -243,8 +245,8 @@ public class StompWebSocketHandler extends TextWebSocketHandlerAdapter implement
}
try {
Message<byte[]> byteMessage = MessageBuilder.fromPayloadAndHeaders(payload,
headers.toMessageHeaders()).build();
@SuppressWarnings("unchecked")
M byteMessage = (M) MessageBuilder.fromPayloadAndHeaders(payload, headers.toMessageHeaders()).build();
byte[] bytes = getStompMessageConverter().fromMessage(byteMessage);
session.sendMessage(new TextMessage(new String(bytes, Charset.forName("UTF-8"))));
}

4
spring-websocket/src/main/java/org/springframework/web/messaging/support/AbstractPubSubChannelRegistry.java
View File

@@ -28,7 +28,9 @@ import org.springframework.web.messaging.PubSubChannelRegistry;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class AbstractPubSubChannelRegistry<M extends Message<?>, H extends MessageHandler<M>> implements PubSubChannelRegistry<M, H>, InitializingBean {
@SuppressWarnings("rawtypes")
public class AbstractPubSubChannelRegistry<M extends Message, H extends MessageHandler<M>>
implements PubSubChannelRegistry<M, H>, InitializingBean {
private SubscribableChannel<M, H> clientInputChannel;

13
spring-websocket/src/main/java/org/springframework/web/messaging/support/ReactorMessageChannel.java
View File

@@ -47,8 +47,8 @@ public class ReactorMessageChannel implements SubscribableChannel<Message<?>, Me
private String name = toString(); // TODO
private final Map<MessageHandler, Registration<?>> registrations =
new HashMap<MessageHandler, Registration<?>>();
private final Map<MessageHandler<Message<?>>, Registration<?>> registrations =
new HashMap<MessageHandler<Message<?>>, Registration<?>>();
public ReactorMessageChannel(Reactor reactor) {
@@ -78,7 +78,7 @@ public class ReactorMessageChannel implements SubscribableChannel<Message<?>, Me
}
@Override
public boolean subscribe(final MessageHandler handler) {
public boolean subscribe(final MessageHandler<Message<?>> handler) {
if (this.registrations.containsKey(handler)) {
logger.warn("Channel " + getName() + ", handler already subscribed " + handler);
@@ -98,7 +98,7 @@ public class ReactorMessageChannel implements SubscribableChannel<Message<?>, Me
}
@Override
public boolean unsubscribe(MessageHandler handler) {
public boolean unsubscribe(MessageHandler<Message<?>> handler) {
if (logger.isTraceEnabled()) {
logger.trace("Channel " + getName() + ", removing subscription for handler " + handler);
@@ -119,13 +119,12 @@ public class ReactorMessageChannel implements SubscribableChannel<Message<?>, Me
private static final class MessageHandlerConsumer implements Consumer<Event<Message<?>>> {
private final MessageHandler handler;
private final MessageHandler<Message<?>> handler;
private MessageHandlerConsumer(MessageHandler handler) {
private MessageHandlerConsumer(MessageHandler<Message<?>> handler) {
this.handler = handler;
}
@SuppressWarnings("unchecked")
@Override
public void accept(Event<Message<?>> event) {
Message<?> message = event.getData();

17
spring-websocket/src/main/java/org/springframework/web/messaging/support/SessionMessageChannel.java
View File

@@ -28,14 +28,15 @@ import reactor.util.Assert;
* @author Rossen Stoyanchev
* @since 4.0
*/
public class SessionMessageChannel implements MessageChannel<Message<?>> {
@SuppressWarnings("rawtypes")
public class SessionMessageChannel<M extends Message> implements MessageChannel<M> {
private MessageChannel<Message<?>> delegate;
private MessageChannel<M> delegate;
private final String sessionId;
public SessionMessageChannel(MessageChannel<Message<?>> delegate, String sessionId) {
public SessionMessageChannel(MessageChannel<M> delegate, String sessionId) {
Assert.notNull(delegate, "delegate is required");
Assert.notNull(sessionId, "sessionId is required");
this.sessionId = sessionId;
@@ -43,17 +44,17 @@ public class SessionMessageChannel implements MessageChannel<Message<?>> {
}
@Override
public boolean send(Message<?> message) {
public boolean send(M message) {
return send(message, -1);
}
@Override
public boolean send(Message<?> message, long timeout) {
public boolean send(M message, long timeout) {
PubSubHeaders headers = PubSubHeaders.fromMessageHeaders(message.getHeaders());
headers.setSessionId(this.sessionId);
MessageBuilder<?> messageToSend = MessageBuilder.fromPayloadAndHeaders(
message.getPayload(), headers.toMessageHeaders());
this.delegate.send(messageToSend.build());
@SuppressWarnings("unchecked")
M messageToSend = (M) MessageBuilder.fromPayloadAndHeaders(message.getPayload(), headers.toMessageHeaders()).build();
this.delegate.send(messageToSend);
return true;
}
}