From 8ce8bf9d7331b479e3a4a1870da69c6aa7e3e0dd Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 19 Dec 2018 12:33:47 -0500 Subject: [PATCH] Fix code smell in websocket and webflux modules (#2669) * * Fix code smell in websocket and webflux modules * * More fixes * * Fix NPE in the `IntegrationHandlerResultHandler` --- .../PayloadTypeConvertingTransformer.java | 2 +- .../DslScriptExecutingMessageProcessor.java | 2 +- .../IntegrationHandlerResultHandler.java | 5 +- .../inbound/WebFluxInboundEndpoint.java | 97 ++++++++++--------- ...tegrationRequestMappingHandlerMapping.java | 4 +- ...WebFluxRequestExecutingMessageHandler.java | 2 +- ...etIntegrationConfigurationInitializer.java | 25 ++--- .../WebSocketInboundChannelAdapter.java | 75 +++++++++----- .../WebSocketOutboundMessageHandler.java | 17 +++- 9 files changed, 138 insertions(+), 91 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java index dd9dc23c47..7b382196a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java @@ -53,7 +53,7 @@ public class PayloadTypeConvertingTransformer extends AbstractPayloadTrans } @Override - protected void onInit() throws Exception { + protected void onInit() throws Exception { // NOSONAR thrown by super class super.onInit(); Assert.notNull(this.converter, () -> getClass().getName() + " requires a Converter"); } diff --git a/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/dsl/DslScriptExecutingMessageProcessor.java b/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/dsl/DslScriptExecutingMessageProcessor.java index bf6597a441..5df34e83c0 100644 --- a/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/dsl/DslScriptExecutingMessageProcessor.java +++ b/spring-integration-scripting/src/main/java/org/springframework/integration/scripting/dsl/DslScriptExecutingMessageProcessor.java @@ -105,7 +105,7 @@ class DslScriptExecutingMessageProcessor String filename = this.script.getFilename(); int index = filename != null - ? filename.lastIndexOf(".") + 1 + ? filename.lastIndexOf('.') + 1 : -1; if (index < 1) { throw new BeanCreationException( diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/IntegrationHandlerResultHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/IntegrationHandlerResultHandler.java index 98ccef31f2..c8b599e43a 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/IntegrationHandlerResultHandler.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/IntegrationHandlerResultHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-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. @@ -47,7 +47,8 @@ public class IntegrationHandlerResultHandler implements HandlerResultHandler, Or @Override @SuppressWarnings("unchecked") public Mono handleResult(ServerWebExchange exchange, HandlerResult result) { - return (Mono) result.getReturnValue(); + Object returnValue = result.getReturnValue(); + return returnValue == null ? Mono.empty() : (Mono) returnValue; } @Override diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java index 5f6aecf504..fa7991a74e 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxInboundEndpoint.java @@ -80,6 +80,8 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W private static final MediaType MEDIA_TYPE_APPLICATION_ALL = new MediaType("application"); + private static final String UNCHECKED = "unchecked"; + private static final List SAFE_METHODS = Arrays.asList(HttpMethod.GET, HttpMethod.HEAD); private ServerCodecConfigurer codecConfigurer = ServerCodecConfigurer.create(); @@ -130,11 +132,6 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W return super.getComponentType().replaceFirst("http", "webflux"); } - @Override - protected void onInit() throws Exception { - super.onInit(); - } - @Override public Mono handle(ServerWebExchange exchange) { return Mono.defer(() -> { @@ -147,7 +144,6 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W }); } - @SuppressWarnings("unchecked") private Mono doHandle(ServerWebExchange exchange) { return extractRequestBody(exchange) .doOnSubscribe(s -> this.activeCount.incrementAndGet()) @@ -170,7 +166,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W } - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) private Mono extractRequestBody(ServerWebExchange exchange) { ServerHttpRequest request = exchange.getRequest(); ServerHttpResponse response = exchange.getResponse(); @@ -201,7 +197,8 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W Class resolvedType = bodyType.resolve(); - ReactiveAdapter adapter = (resolvedType != null ? this.adapterRegistry.getAdapter(resolvedType) : null); + ReactiveAdapter adapter = (resolvedType != null ? this.adapterRegistry.getAdapter(resolvedType) : + null); ResolvableType elementType = (adapter != null ? bodyType.getGeneric() : bodyType); HttpMessageReader httpMessageReader = this.codecConfigurer @@ -237,40 +234,14 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W } } - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) private Mono, RequestEntity>> buildMessage(RequestEntity httpEntity, ServerWebExchange exchange) { ServerHttpRequest request = exchange.getRequest(); - HttpHeaders requestHeaders = request.getHeaders(); - Map exchangeAttributes = exchange.getAttributes(); - - StandardEvaluationContext evaluationContext = createEvaluationContext(); - - evaluationContext.setVariable("requestAttributes", exchangeAttributes); MultiValueMap requestParams = request.getQueryParams(); - evaluationContext.setVariable("requestParams", requestParams); - evaluationContext.setVariable("requestHeaders", requestHeaders); - if (!CollectionUtils.isEmpty(request.getCookies())) { - evaluationContext.setVariable("cookies", request.getCookies()); - } - Map pathVariables = - (Map) exchangeAttributes.get(HandlerMapping.URI_TEMPLATE_VARIABLES_ATTRIBUTE); - - if (!CollectionUtils.isEmpty(pathVariables)) { - evaluationContext.setVariable("pathVariables", pathVariables); - } - - Map> matrixVariables = - (Map>) exchangeAttributes - .get(HandlerMapping.MATRIX_VARIABLES_ATTRIBUTE); - - if (!CollectionUtils.isEmpty(matrixVariables)) { - evaluationContext.setVariable("matrixVariables", matrixVariables); - } - - evaluationContext.setRootObject(httpEntity); + StandardEvaluationContext evaluationContext = buildEvaluationContext(httpEntity, exchange); Object payload; if (getPayloadExpression() != null) { payload = getPayloadExpression().getValue(evaluationContext); @@ -310,10 +281,13 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W .copyHeaders(headers); } - messageBuilder - .setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_URL, request.getURI().toString()) - .setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD, - request.getMethod().toString()); + messageBuilder.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_URL, + request.getURI().toString()); + HttpMethod httpMethod = request.getMethod(); + if (httpMethod != null) { + messageBuilder.setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD, + httpMethod.toString()); + } return exchange.getPrincipal() .map(principal -> @@ -324,6 +298,41 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W .zipWith(Mono.just(httpEntity)); } + @SuppressWarnings(UNCHECKED) + private StandardEvaluationContext buildEvaluationContext(RequestEntity httpEntity, ServerWebExchange exchange) { + ServerHttpRequest request = exchange.getRequest(); + HttpHeaders requestHeaders = request.getHeaders(); + MultiValueMap requestParams = request.getQueryParams(); + Map exchangeAttributes = exchange.getAttributes(); + + StandardEvaluationContext evaluationContext = createEvaluationContext(); + + evaluationContext.setVariable("requestAttributes", exchangeAttributes); + evaluationContext.setVariable("requestParams", requestParams); + evaluationContext.setVariable("requestHeaders", requestHeaders); + if (!CollectionUtils.isEmpty(request.getCookies())) { + evaluationContext.setVariable("cookies", request.getCookies()); + } + + Map pathVariables = + (Map) exchangeAttributes.get(HandlerMapping.URI_TEMPLATE_VARIABLES_ATTRIBUTE); + + if (!CollectionUtils.isEmpty(pathVariables)) { + evaluationContext.setVariable("pathVariables", pathVariables); + } + + Map> matrixVariables = + (Map>) exchangeAttributes + .get(HandlerMapping.MATRIX_VARIABLES_ATTRIBUTE); + + if (!CollectionUtils.isEmpty(matrixVariables)) { + evaluationContext.setVariable("matrixVariables", matrixVariables); + } + + evaluationContext.setRootObject(httpEntity); + return evaluationContext; + } + private Mono populateResponse(ServerWebExchange exchange, Message replyMessage) { ServerHttpResponse response = exchange.getResponse(); getHeaderMapper().fromHeaders(replyMessage.getHeaders(), response.getHeaders()); @@ -379,7 +388,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W } } - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) private Mono writeResponseBody(ServerWebExchange exchange, Object body) { ResolvableType bodyType = ResolvableType.forInstance(body); ReactiveAdapter adapter = this.adapterRegistry.getAdapter(bodyType.resolve(), body); @@ -473,7 +482,7 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W return (mediaTypes.isEmpty() ? Collections.singletonList(MediaType.ALL) : mediaTypes); } - @SuppressWarnings("unchecked") + @SuppressWarnings(UNCHECKED) private List getProducibleTypes(ServerWebExchange exchange, Supplier> producibleTypesSupplier) { @@ -482,9 +491,9 @@ public class WebFluxInboundEndpoint extends BaseHttpInboundEndpoint implements W } private MediaType selectMoreSpecificMediaType(MediaType acceptable, MediaType producible) { - producible = producible.copyQualityValue(acceptable); + MediaType producibleToUse = producible.copyQualityValue(acceptable); Comparator comparator = MediaType.SPECIFICITY_COMPARATOR; - return (comparator.compare(acceptable, producible) <= 0 ? acceptable : producible); + return (comparator.compare(acceptable, producibleToUse) <= 0 ? acceptable : producibleToUse); } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java index d9540932e0..1b214fa8f2 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/inbound/WebFluxIntegrationRequestMappingHandlerMapping.java @@ -119,11 +119,11 @@ public class WebFluxIntegrationRequestMappingHandlerMapping extends RequestMappi @Override protected void detectHandlerMethods(Object handler) { if (handler instanceof String) { - handler = getApplicationContext().getBean((String) handler); + handler = getApplicationContext().getBean((String) handler); // NOSONAR never null } RequestMappingInfo mapping = getMappingForEndpoint((WebFluxInboundEndpoint) handler); if (mapping != null) { - registerMapping(mapping, handler, HANDLER_METHOD); + registerMapping(mapping, handler, HANDLER_METHOD); // NOSONAR never null } } diff --git a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java index 55ed5f3699..1ddbf5c3da 100644 --- a/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java +++ b/spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java @@ -159,7 +159,7 @@ public class WebFluxRequestExecutingMessageHandler extends AbstractHttpRequestEx .headers(headers -> headers.putAll(httpRequest.getHeaders())); if (httpRequest.hasBody()) { - requestSpec.body(BodyInserters.fromObject(httpRequest.getBody())); + requestSpec.body(BodyInserters.fromObject(httpRequest.getBody())); // NOSONAR protected with hasBody() } Mono responseMono = diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketIntegrationConfigurationInitializer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketIntegrationConfigurationInitializer.java index 6f9672683a..7cfb6040be 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketIntegrationConfigurationInitializer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketIntegrationConfigurationInitializer.java @@ -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 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); diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java index 823c436220..e10f38517c 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/inbound/WebSocketInboundChannelAdapter.java @@ -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 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) message)); @@ -306,9 +298,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport this.eventPublisher.publishEvent(new ReceiptEvent(this, (Message) 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 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 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 messageToSend = + getMessageBuilderFactory() + .withPayload(payload) + .copyHeaders(headerAccessor.toMap()) + .build(); + sendMessage(messageToSend); + } + } diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/outbound/WebSocketOutboundMessageHandler.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/outbound/WebSocketOutboundMessageHandler.java index 0d83c5dc94..f2ed18b6b7 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/outbound/WebSocketOutboundMessageHandler.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/outbound/WebSocketOutboundMessageHandler.java @@ -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 defaultConverters = new ArrayList(3); + private final List 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 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 messageConverters) { Assert.noNullElements(messageConverters.toArray(), "'messageConverters' must not contain null entries"); - this.messageConverters = new ArrayList(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) {