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

Fix code smell in websocket and webflux modules (#2669)

Browse Source 
* * Fix code smell in websocket and webflux modules

* * More fixes

* * Fix NPE in the `IntegrationHandlerResultHandler`
This commit is contained in:
Artem Bilan
2018-12-19 12:33:47 -05:00
committed by Gary Russell
parent e8bd31cc37
commit 8ce8bf9d73
9 changed files with 138 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketIntegrationConfigurationInitializer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-2018 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,12 +16,11 @@
package org.springframework.integration.websocket.config;
import java.util.Collection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.beans.factory.config.BeanDefinition;
@@ -50,10 +49,12 @@ import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
*/
public class WebSocketIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer {
private static final Log logger = LogFactory.getLog(WebSocketIntegrationConfigurationInitializer.class);
private static final Log logger = // NOSONAR lower case
LogFactory.getLog(WebSocketIntegrationConfigurationInitializer.class);
private static final boolean servletPresent = ClassUtils.isPresent("javax.servlet.Servlet",
WebSocketIntegrationConfigurationInitializer.class.getClassLoader());
private static final boolean servletPresent = // NOSONAR lower case
ClassUtils.isPresent("javax.servlet.Servlet",
WebSocketIntegrationConfigurationInitializer.class.getClassLoader());
private static final String WEB_SOCKET_HANDLER_MAPPING_BEAN_NAME = "integrationWebSocketHandlerMapping";
@@ -130,11 +131,13 @@ public class WebSocketIntegrationConfigurationInitializer implements Integration
}
@Override
protected HandlerMapping createInstance() throws Exception {
Collection<WebSocketConfigurer> webSocketConfigurers =
((ListableBeanFactory) getBeanFactory()).getBeansOfType(WebSocketConfigurer.class).values();
for (WebSocketConfigurer configurer : webSocketConfigurers) {
configurer.registerWebSocketHandlers(this.registry);
protected HandlerMapping createInstance() {
BeanFactory beanFactory = getBeanFactory();
if (beanFactory != null) {
((ListableBeanFactory) beanFactory)
.getBeansOfType(WebSocketConfigurer.class)
.values()
.forEach(configurer -> configurer.registerWebSocketHandlers(this.registry));
}
if (this.registry.requiresTaskScheduler()) {
this.registry.setTaskScheduler(this.sockJsTaskScheduler);

75
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java
View File

@@ -220,7 +220,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
}
@Override
public void afterSessionStarted(WebSocketSession session) throws Exception {
public void afterSessionStarted(WebSocketSession session) throws Exception { // NOSONAR Thrown from the delegate
if (isActive()) {
SubProtocolHandler protocolHandler = this.subProtocolHandlerRegistry.findProtocolHandler(session);
protocolHandler.afterSessionStarted(session, this.subProtocolHandlerChannel);
@@ -237,7 +237,8 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
}
@Override
public void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus) throws Exception {
public void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus)
throws Exception { // NOSONAR Thrown from the delegate
if (isActive()) {
this.subProtocolHandlerRegistry.findProtocolHandler(session)
.afterSessionEnded(session, closeStatus, this.subProtocolHandlerChannel);
@@ -245,7 +246,8 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
}
@Override
public void onMessage(WebSocketSession session, WebSocketMessage<?> webSocketMessage) throws Exception {
public void onMessage(WebSocketSession session, WebSocketMessage<?> webSocketMessage)
throws Exception { // NOSONAR Thrown from the delegate
if (isActive()) {
this.subProtocolHandlerRegistry.findProtocolHandler(session)
.handleMessageFromClient(session, webSocketMessage, this.subProtocolHandlerChannel);
@@ -281,23 +283,13 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
}
@SuppressWarnings("unchecked")
private void handleMessageAndSend(Message<?> message) throws Exception {
private void handleMessageAndSend(final Message<?> message) {
SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message);
StompCommand stompCommand = (StompCommand) headerAccessor.getHeader("stompCommand");
SimpMessageType messageType = headerAccessor.getMessageType();
if ((messageType == null || SimpMessageType.MESSAGE.equals(messageType)
|| (SimpMessageType.CONNECT.equals(messageType) && !this.useBroker)
|| StompCommand.CONNECTED.equals(stompCommand)
|| StompCommand.RECEIPT.equals(stompCommand))
&& !checkDestinationPrefix(headerAccessor.getDestination())) {
if (isProcessingTypeOrCommand(headerAccessor, stompCommand, messageType)) {
if (SimpMessageType.CONNECT.equals(messageType)) {
String sessionId = headerAccessor.getSessionId();
SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
connectAck.setSessionId(sessionId);
connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> ackMessage = MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());
WebSocketSession session = this.webSocketContainer.getSession(sessionId);
this.subProtocolHandlerRegistry.findProtocolHandler(session).handleMessageToClient(session, ackMessage);
produceConnectAckMessage(message, headerAccessor);
}
else if (StompCommand.CONNECTED.equals(stompCommand)) {
this.eventPublisher.publishEvent(new SessionConnectedEvent(this, (Message<byte[]>) message));
@@ -306,9 +298,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
this.eventPublisher.publishEvent(new ReceiptEvent(this, (Message<byte[]>) message));
}
else {
headerAccessor.removeHeader(SimpMessageHeaderAccessor.NATIVE_HEADERS);
Object payload = this.messageConverter.fromMessage(message, this.payloadType.get());
sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headerAccessor.toMap()).build());
produceMessage(message, headerAccessor);
}
}
else {
@@ -324,19 +314,56 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport
}
}
private boolean isProcessingTypeOrCommand(SimpMessageHeaderAccessor headerAccessor, StompCommand stompCommand,
SimpMessageType messageType) {
return (messageType == null // NOSONAR pretty simple logic
|| SimpMessageType.MESSAGE.equals(messageType)
|| (SimpMessageType.CONNECT.equals(messageType) && !this.useBroker)
|| StompCommand.CONNECTED.equals(stompCommand)
|| StompCommand.RECEIPT.equals(stompCommand))
&& !checkDestinationPrefix(headerAccessor.getDestination());
}
private boolean checkDestinationPrefix(String destination) {
if (this.useBroker) {
Collection<String> destinationPrefixes = this.brokerHandler.getDestinationPrefixes();
if ((destination == null) || CollectionUtils.isEmpty(destinationPrefixes)) {
return false;
}
for (String prefix : destinationPrefixes) {
if (destination.startsWith(prefix)) {
return true;
}
}
return destinationPrefixes.stream().anyMatch(destination::startsWith);
}
return false;
}
private void produceConnectAckMessage(Message<?> message, SimpMessageHeaderAccessor headerAccessor) {
String sessionId = headerAccessor.getSessionId();
SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
connectAck.setSessionId(sessionId);
connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
Message<byte[]> ackMessage = MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());
WebSocketSession session = this.webSocketContainer.getSession(sessionId);
try {
this.subProtocolHandlerRegistry.findProtocolHandler(session).handleMessageToClient(session, ackMessage);
}
catch (Exception e) {
throw new MessageHandlingException(message, "Error sending connect ack message", e);
}
}
private void produceMessage(Message<?> message, SimpMessageHeaderAccessor headerAccessor) {
headerAccessor.removeHeader(SimpMessageHeaderAccessor.NATIVE_HEADERS);
Object payload = this.messageConverter.fromMessage(message, this.payloadType.get());
Assert.state(payload != null,
() -> "The message converter '" + this.messageConverter +
"' produced no payload for message '" + message +
"' and expected payload type: " + this.payloadType.get());
Message<Object> messageToSend =
getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(headerAccessor.toMap())
.build();
sendMessage(messageToSend);
}
}

17
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/outbound/WebSocketOutboundMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-2018 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.
@@ -43,11 +43,12 @@ import org.springframework.web.socket.handler.SessionLimitExceededException;
/**
* @author Artem Bilan
*
* @since 4.1
*/
public class WebSocketOutboundMessageHandler extends AbstractMessageHandler {
private final List<MessageConverter> defaultConverters = new ArrayList<MessageConverter>(3);
private final List<MessageConverter> defaultConverters = new ArrayList<>(3);
{
this.defaultConverters.add(new StringMessageConverter());
@@ -79,13 +80,14 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler {
public WebSocketOutboundMessageHandler(IntegrationWebSocketContainer webSocketContainer,
SubProtocolHandlerRegistry protocolHandlerRegistry) {
Assert.notNull(webSocketContainer, "'webSocketContainer' must not be null");
Assert.notNull(protocolHandlerRegistry, "'protocolHandlerRegistry' must not be null");
this.webSocketContainer = webSocketContainer;
this.client = webSocketContainer instanceof ClientWebSocketContainer;
this.subProtocolHandlerRegistry = protocolHandlerRegistry;
List<String> subProtocols = protocolHandlerRegistry.getSubProtocols();
this.webSocketContainer.addSupportedProtocols(subProtocols.toArray(new String[subProtocols.size()]));
this.webSocketContainer.addSupportedProtocols(subProtocols.toArray(new String[0]));
}
/**
@@ -95,7 +97,7 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler {
*/
public void setMessageConverters(List<MessageConverter> messageConverters) {
Assert.noNullElements(messageConverters.toArray(), "'messageConverters' must not contain null entries");
this.messageConverters = new ArrayList<MessageConverter>(messageConverters);
this.messageConverters = new ArrayList<>(messageConverters);
}
@@ -147,7 +149,12 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler {
SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
headers.setLeaveMutable(true);
headers.setMessageTypeIfNotSet(SimpMessageType.MESSAGE);
Message<?> messageToSend = this.messageConverter.toMessage(message.getPayload(), headers.getMessageHeaders());
Object payload = message.getPayload();
Message<?> messageToSend =
this.messageConverter.toMessage(payload, headers.getMessageHeaders());
Assert.state(messageToSend != null,
() -> "The message converter '" + this.messageConverter +
"' produced no message to send based on the request message: '" + message + "'");
this.subProtocolHandlerRegistry.findProtocolHandler(session).handleMessageToClient(session, messageToSend);
}
catch (SessionLimitExceededException ex) {