diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/JavaUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/JavaUtils.java new file mode 100644 index 0000000000..2d6329dd29 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/JavaUtils.java @@ -0,0 +1,70 @@ +/* + * Copyright 2019 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.integration.util; + +import java.util.function.Consumer; + +/** + * Chained utility methods to simplify some Java repetitive code. Obtain a reference to + * the singleton {@link #INSTANCE} and then chain calls to the utility methods. + * + * @author Gary Russell + * @author Artem Bilan + * + * @since 5.1.3 + */ +public final class JavaUtils { + + /** + * The singleton instance of this utility class. + */ + public static final JavaUtils INSTANCE = new JavaUtils(); + + private JavaUtils() { + super(); + } + + /** + * Invoke {@link Consumer#accept(Object)} with the value if the condition is true. + * @param condition the condition. + * @param value the value. + * @param consumer the consumer. + * @param the value type. + * @return this. + */ + public JavaUtils acceptIfCondition(boolean condition, T value, Consumer consumer) { + if (condition) { + consumer.accept(value); + } + return this; + } + + /** + * Invoke {@link Consumer#accept(Object)} with the value if it is not null. + * @param value the value. + * @param consumer the consumer. + * @param the value type. + * @return this. + */ + public JavaUtils acceptIfNotNull(T value, Consumer consumer) { + if (value != null) { + consumer.accept(value); + } + return this; + } + +} diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java index dfcf2ee2f8..7a25107500 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2019 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. @@ -46,6 +46,7 @@ import org.springframework.web.socket.client.WebSocketClient; * * @author Artem Bilan * @author Gary Russell + * * @since 4.1 */ public final class ClientWebSocketContainer extends IntegrationWebSocketContainer implements SmartLifecycle { @@ -221,7 +222,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine } @Override - public void stopInternal() throws Exception { + public void stopInternal() throws Exception { // NOSONAR honor super if (this.syncClientLifecycle) { ((Lifecycle) this.client).stop(); } @@ -236,7 +237,9 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine @Override protected void openConnection() { - logger.info("Connecting to WebSocket at " + getUri()); + if (logger.isInfoEnabled()) { + logger.info("Connecting to WebSocket at " + getUri()); + } ClientWebSocketContainer.this.headers.setSecWebSocketProtocol(getSubProtocols()); ListenableFuture future = this.client.doHandshake(ClientWebSocketContainer.this.webSocketHandler, @@ -262,10 +265,9 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine } @Override - protected void closeConnection() throws Exception { + protected void closeConnection() throws Exception { // NOSONAR if (ClientWebSocketContainer.this.clientSession != null) { - ClientWebSocketContainer.this.closeSession(ClientWebSocketContainer.this.clientSession, - CloseStatus.NORMAL); + closeSession(ClientWebSocketContainer.this.clientSession, CloseStatus.NORMAL); } } diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java index c84e35e529..db257b88e1 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/IntegrationWebSocketContainer.java @@ -27,6 +27,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.DisposableBean; import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.SubProtocolCapable; import org.springframework.web.socket.WebSocketHandler; @@ -52,25 +53,31 @@ import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorato * * @author Artem Bilan * @author Gary Russell + * * @since 4.1 + * * @see org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter * @see org.springframework.integration.websocket.outbound.WebSocketOutboundMessageHandler */ public abstract class IntegrationWebSocketContainer implements DisposableBean { - protected final Log logger = LogFactory.getLog(this.getClass()); + public static final int DEFAULT_SEND_TIME_LIMIT = 10 * 1000; - protected final WebSocketHandler webSocketHandler = new IntegrationWebSocketHandler(); + public static final int DEFAULT_SEND_BUFFER_SIZE = 512 * 1024; - protected final Map sessions = new ConcurrentHashMap(); + protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR - private final List supportedProtocols = new ArrayList(); + protected final WebSocketHandler webSocketHandler = new IntegrationWebSocketHandler(); // NOSONAR + + protected final Map sessions = new ConcurrentHashMap<>(); // NOSONAR + + private final List supportedProtocols = new ArrayList<>(); private volatile WebSocketListener messageListener; - private volatile int sendTimeLimit = 10 * 1000; + private volatile int sendTimeLimit = DEFAULT_SEND_TIME_LIMIT; - private volatile int sendBufferSizeLimit = 512 * 1024; + private volatile int sendBufferSizeLimit = DEFAULT_SEND_BUFFER_SIZE; public void setSendTimeLimit(int sendTimeLimit) { this.sendTimeLimit = sendTimeLimit; @@ -98,7 +105,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { } public List getSubProtocols() { - List protocols = new ArrayList(); + List protocols = new ArrayList<>(); if (this.messageListener != null) { protocols.addAll(this.messageListener.getSubProtocols()); } @@ -116,14 +123,16 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { return session; } - public void closeSession(WebSocketSession session, CloseStatus closeStatus) throws Exception { + public void closeSession(WebSocketSession session, CloseStatus closeStatus) + throws Exception { // NOSONAR + // Session may be unresponsive so clear first session.close(closeStatus); this.webSocketHandler.afterConnectionClosed(session, closeStatus); } @Override - public void destroy() throws Exception { + public void destroy() { try { // Notify sessions to stop flushing messages for (WebSocketSession session : this.sessions.values()) { @@ -159,15 +168,19 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { } @Override - public void afterConnectionEstablished(WebSocketSession sessionToDecorate) throws Exception { // NOSONAR SF ifce - WebSocketSession session = new ConcurrentWebSocketSessionDecorator(sessionToDecorate, - IntegrationWebSocketContainer.this.sendTimeLimit, - IntegrationWebSocketContainer.this.sendBufferSizeLimit); + public void afterConnectionEstablished(WebSocketSession sessionToDecorate) + throws Exception { // NOSONAR + + WebSocketSession session = + new ConcurrentWebSocketSessionDecorator(sessionToDecorate, + IntegrationWebSocketContainer.this.sendTimeLimit, + IntegrationWebSocketContainer.this.sendBufferSizeLimit); IntegrationWebSocketContainer.this.sessions.put(session.getId(), session); if (IntegrationWebSocketContainer.this.logger.isDebugEnabled()) { - IntegrationWebSocketContainer.this.logger.debug("Started WebSocket session = " + session.getId() + ", number of sessions = " - + IntegrationWebSocketContainer.this.sessions.size()); + IntegrationWebSocketContainer.this.logger.debug("Started WebSocket session = " + + session.getId() + ", number of sessions = " + + IntegrationWebSocketContainer.this.sessions.size()); } if (IntegrationWebSocketContainer.this.messageListener != null) { IntegrationWebSocketContainer.this.messageListener.afterSessionStarted(session); @@ -175,29 +188,34 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean { } @Override - public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception { + public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) + throws Exception { // NOSONAR + WebSocketSession removed = IntegrationWebSocketContainer.this.sessions.remove(session.getId()); - if (removed != null) { - if (IntegrationWebSocketContainer.this.messageListener != null) { - IntegrationWebSocketContainer.this.messageListener.afterSessionEnded(session, closeStatus); - } + if (removed != null && IntegrationWebSocketContainer.this.messageListener != null) { + IntegrationWebSocketContainer.this.messageListener.afterSessionEnded(session, closeStatus); } } @Override - public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { + public void handleTransportError(WebSocketSession session, Throwable exception) + throws Exception { // NOSONAR + IntegrationWebSocketContainer.this.sessions.remove(session.getId()); - throw new Exception(exception); + ReflectionUtils.rethrowException(exception); } @Override - public void handleMessage(WebSocketSession session, WebSocketMessage message) throws Exception { + public void handleMessage(WebSocketSession session, WebSocketMessage message) + throws Exception { // NOSONAR + if (IntegrationWebSocketContainer.this.messageListener != null) { IntegrationWebSocketContainer.this.messageListener.onMessage(session, message); } else if (IntegrationWebSocketContainer.this.logger.isInfoEnabled()) { - IntegrationWebSocketContainer.this.logger.info("This 'WebSocketHandlerContainer' isn't configured with 'WebSocketMessageListener'." - + " Received messages are ignored. Current message is: " + message); + IntegrationWebSocketContainer.this.logger.info("This 'WebSocketHandlerContainer' isn't " + + "configured with 'WebSocketMessageListener'." + + " Received messages are ignored. Current message is: " + message); } } diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java index 08811bf3ca..e6dec92d5b 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2019 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. @@ -20,6 +20,7 @@ import java.util.Arrays; import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; +import org.springframework.integration.util.JavaUtils; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -46,6 +47,7 @@ import org.springframework.web.socket.sockjs.transport.TransportHandler; * * @author Artem Bilan * @author Gary Russell + * * @since 4.1 */ public class ServerWebSocketContainer extends IntegrationWebSocketContainer @@ -142,47 +144,38 @@ public class ServerWebSocketContainer extends IntegrationWebSocketContainer .addInterceptors(this.interceptors) .setAllowedOrigins(this.origins); + configureSockJsOptionsIfAny(registration); + } + + private void configureSockJsOptionsIfAny(WebSocketHandlerRegistration registration) { if (this.sockJsServiceOptions != null) { SockJsServiceRegistration sockJsServiceRegistration = registration.withSockJS(); - if (this.sockJsServiceOptions.webSocketEnabled != null) { - sockJsServiceRegistration.setWebSocketEnabled(this.sockJsServiceOptions.webSocketEnabled); - } - if (this.sockJsServiceOptions.clientLibraryUrl != null) { - sockJsServiceRegistration.setClientLibraryUrl(this.sockJsServiceOptions.clientLibraryUrl); - } - if (this.sockJsServiceOptions.disconnectDelay != null) { - sockJsServiceRegistration.setDisconnectDelay(this.sockJsServiceOptions.disconnectDelay); - } - if (this.sockJsServiceOptions.heartbeatTime != null) { - sockJsServiceRegistration.setHeartbeatTime(this.sockJsServiceOptions.heartbeatTime); - } - if (this.sockJsServiceOptions.httpMessageCacheSize != null) { - sockJsServiceRegistration.setHttpMessageCacheSize(this.sockJsServiceOptions.httpMessageCacheSize); - } - if (this.sockJsServiceOptions.heartbeatTime != null) { - sockJsServiceRegistration.setHeartbeatTime(this.sockJsServiceOptions.heartbeatTime); - } - if (this.sockJsServiceOptions.sessionCookieNeeded != null) { - sockJsServiceRegistration.setSessionCookieNeeded(this.sockJsServiceOptions.sessionCookieNeeded); - } - if (this.sockJsServiceOptions.streamBytesLimit != null) { - sockJsServiceRegistration.setStreamBytesLimit(this.sockJsServiceOptions.streamBytesLimit); - } - if (this.sockJsServiceOptions.transportHandlers != null) { - sockJsServiceRegistration.setTransportHandlers(this.sockJsServiceOptions.transportHandlers); - } - if (this.sockJsServiceOptions.taskScheduler != null) { - sockJsServiceRegistration.setTaskScheduler(this.sockJsServiceOptions.taskScheduler); - } - if (this.sockJsServiceOptions.messageCodec != null) { - sockJsServiceRegistration.setMessageCodec(this.sockJsServiceOptions.messageCodec); - } - if (this.sockJsServiceOptions.suppressCors != null) { - sockJsServiceRegistration.setSupressCors(this.sockJsServiceOptions.suppressCors); - } - + JavaUtils.INSTANCE + .acceptIfNotNull(this.sockJsServiceOptions.webSocketEnabled, + sockJsServiceRegistration::setWebSocketEnabled) + .acceptIfNotNull(this.sockJsServiceOptions.clientLibraryUrl, + sockJsServiceRegistration::setClientLibraryUrl) + .acceptIfNotNull(this.sockJsServiceOptions.disconnectDelay, + sockJsServiceRegistration::setDisconnectDelay) + .acceptIfNotNull(this.sockJsServiceOptions.heartbeatTime, + sockJsServiceRegistration::setHeartbeatTime) + .acceptIfNotNull(this.sockJsServiceOptions.httpMessageCacheSize, + sockJsServiceRegistration::setHttpMessageCacheSize) + .acceptIfNotNull(this.sockJsServiceOptions.heartbeatTime, + sockJsServiceRegistration::setHeartbeatTime) + .acceptIfNotNull(this.sockJsServiceOptions.sessionCookieNeeded, + sockJsServiceRegistration::setSessionCookieNeeded) + .acceptIfNotNull(this.sockJsServiceOptions.streamBytesLimit, + sockJsServiceRegistration::setStreamBytesLimit) + .acceptIfNotNull(this.sockJsServiceOptions.transportHandlers, + sockJsServiceRegistration::setTransportHandlers) + .acceptIfNotNull(this.sockJsServiceOptions.taskScheduler, + sockJsServiceRegistration::setTaskScheduler) + .acceptIfNotNull(this.sockJsServiceOptions.messageCodec, + sockJsServiceRegistration::setMessageCodec) + .acceptIfNotNull(this.sockJsServiceOptions.suppressCors, + sockJsServiceRegistration::setSupressCors); } - } public void setAutoStartup(boolean autoStartup) { diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/WebSocketListener.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/WebSocketListener.java index 488759d554..2dc30019f8 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/WebSocketListener.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/WebSocketListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2019 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. @@ -31,7 +31,9 @@ import org.springframework.web.socket.WebSocketSession; * * @author Andy Wilkinson * @author Artem Bilan + * * @since 4.1 + * * @see org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter */ public interface WebSocketListener extends SubProtocolCapable { @@ -42,14 +44,15 @@ public interface WebSocketListener extends SubProtocolCapable { * @param message the WebSocket message * @throws Exception the 'onMessage' Exception */ - void onMessage(WebSocketSession session, WebSocketMessage message) throws Exception; + void onMessage(WebSocketSession session, WebSocketMessage message) + throws Exception; // NOSONAR Remove in 5.2 /** * Invoked after a {@link WebSocketSession} has started. * @param session the WebSocket session * @throws Exception the 'afterSessionStarted' Exception */ - void afterSessionStarted(WebSocketSession session) throws Exception; + void afterSessionStarted(WebSocketSession session) throws Exception; // NOSONAR Remove in 5.2 /** * Invoked after a {@link WebSocketSession} has ended. @@ -57,6 +60,7 @@ public interface WebSocketListener extends SubProtocolCapable { * @param closeStatus the reason why the session was closed * @throws Exception the 'afterSessionEnded' Exception */ - void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus) throws Exception; + void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus) + throws Exception; // NOSONAR Remove in 5.2 } 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 7cfb6040be..4a140f87fb 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-2018 the original author or authors. + * Copyright 2014-2019 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. @@ -49,10 +49,9 @@ import org.springframework.web.socket.config.annotation.WebSocketConfigurer; */ public class WebSocketIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer { - private static final Log logger = // NOSONAR lower case - LogFactory.getLog(WebSocketIntegrationConfigurationInitializer.class); + private static final Log LOGGER = LogFactory.getLog(WebSocketIntegrationConfigurationInitializer.class); - private static final boolean servletPresent = // NOSONAR lower case + private static final boolean SERVLET_PRESENT = ClassUtils.isPresent("javax.servlet.Servlet", WebSocketIntegrationConfigurationInitializer.class.getClassLoader()); @@ -64,7 +63,7 @@ public class WebSocketIntegrationConfigurationInitializer implements Integration this.registerEnableWebSocketIfNecessary((BeanDefinitionRegistry) beanFactory); } else { - logger.warn("'DelegatingWebSocketConfiguration' isn't registered because 'beanFactory'" + + LOGGER.warn("'DelegatingWebSocketConfiguration' isn't registered because 'beanFactory'" + " isn't an instance of `BeanDefinitionRegistry`."); } } @@ -86,7 +85,7 @@ public class WebSocketIntegrationConfigurationInitializer implements Integration * applications when {@link org.springframework.integration.config.EnableIntegration} is in use. */ private void registerEnableWebSocketIfNecessary(BeanDefinitionRegistry registry) { - if (servletPresent) { + if (SERVLET_PRESENT) { if (!registry.containsBeanDefinition("defaultSockJsTaskScheduler")) { BeanDefinitionBuilder sockJsTaskSchedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(ThreadPoolTaskScheduler.class) @@ -156,13 +155,17 @@ public class WebSocketIntegrationConfigurationInitializer implements Integration private static class IntegrationServletWebSocketHandlerRegistry extends ServletWebSocketHandlerRegistry { + IntegrationServletWebSocketHandlerRegistry() { + super(); + } + @Override - public boolean requiresTaskScheduler() { + public boolean requiresTaskScheduler() { // NOSONAR visibility return super.requiresTaskScheduler(); } @Override - public void setTaskScheduler(TaskScheduler scheduler) { + public void setTaskScheduler(TaskScheduler scheduler) { // NOSONAR visibility super.setTaskScheduler(scheduler); } 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 7634ed4841..41dcc03447 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-2018 the original author or authors. + * Copyright 2014-2019 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. @@ -136,7 +136,7 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler { } @Override - protected void handleMessageInternal(Message message) throws Exception { + protected void handleMessageInternal(Message message) throws Exception { // NOSONAR String sessionId = null; if (!this.client) { sessionId = this.subProtocolHandlerRegistry.resolveSessionId(message); diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/PassThruSubProtocolHandler.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/PassThruSubProtocolHandler.java index af125d1b15..8689586377 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/PassThruSubProtocolHandler.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/PassThruSubProtocolHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2019 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. @@ -41,11 +41,12 @@ import org.springframework.web.socket.messaging.SubProtocolHandler; * on 'send' part and vise versa - on 'receive' part. * * @author Artem Bilan + * * @since 4.1 */ public class PassThruSubProtocolHandler implements SubProtocolHandler { - final List supportedProtocols = new ArrayList(); + private final List supportedProtocols = new ArrayList<>(); public void setSupportedProtocols(String... supportedProtocols) { Assert.noNullElements(supportedProtocols, "'supportedProtocols' must not be empty"); @@ -59,7 +60,8 @@ public class PassThruSubProtocolHandler implements SubProtocolHandler { @Override public void handleMessageFromClient(WebSocketSession session, WebSocketMessage webSocketMessage, - MessageChannel outputChannel) throws Exception { + MessageChannel outputChannel) { + SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE); headerAccessor.setSessionId(session.getId()); headerAccessor.setSessionAttributes(session.getAttributes()); @@ -78,7 +80,9 @@ public class PassThruSubProtocolHandler implements SubProtocolHandler { } @Override - public void handleMessageToClient(WebSocketSession session, Message message) throws Exception { + public void handleMessageToClient(WebSocketSession session, Message message) + throws Exception { // NOSONAR + Object payload = message.getPayload(); if (payload instanceof String) { session.sendMessage(new TextMessage((String) payload)); @@ -91,7 +95,7 @@ public class PassThruSubProtocolHandler implements SubProtocolHandler { } else { throw new IllegalArgumentException("Unsupported payload type: " + payload.getClass() - + ". Can be one of: " + Arrays.>asList(String.class, byte[].class, ByteBuffer.class)); + + ". Can be one of: " + Arrays.asList(String.class, byte[].class, ByteBuffer.class)); } } @@ -101,13 +105,12 @@ public class PassThruSubProtocolHandler implements SubProtocolHandler { } @Override - public void afterSessionStarted(WebSocketSession session, MessageChannel outputChannel) throws Exception { + public void afterSessionStarted(WebSocketSession session, MessageChannel outputChannel) { // Subclasses might implement this method } @Override - public void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus, MessageChannel outputChannel) - throws Exception { + public void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus, MessageChannel outputChannel) { // Subclasses might implement this method } diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/SubProtocolHandlerRegistry.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/SubProtocolHandlerRegistry.java index 0b060f5bff..5c13dd630f 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/SubProtocolHandlerRegistry.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/support/SubProtocolHandlerRegistry.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2018 the original author or authors. + * Copyright 2014-2019 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. @@ -62,9 +62,25 @@ public final class SubProtocolHandlerRegistry { public SubProtocolHandlerRegistry(List protocolHandlers, SubProtocolHandler defaultProtocolHandler) { + Assert.state(!CollectionUtils.isEmpty(protocolHandlers) || defaultProtocolHandler != null, "One of 'protocolHandlers' or 'defaultProtocolHandler' must be provided"); + configureProtocolHandlers(protocolHandlers); + + if (this.protocolHandlers.size() == 1 && defaultProtocolHandler == null) { + this.defaultProtocolHandler = this.protocolHandlers.values().iterator().next(); + } + else { + this.defaultProtocolHandler = defaultProtocolHandler; + if (this.protocolHandlers.isEmpty() && this.defaultProtocolHandler != null) { + List protocols = this.defaultProtocolHandler.getSupportedProtocols(); + populateProtocolsForHandler(this.defaultProtocolHandler, protocols); + } + } + } + + private void configureProtocolHandlers(List protocolHandlers) { if (!CollectionUtils.isEmpty(protocolHandlers)) { for (SubProtocolHandler handler : protocolHandlers) { List protocols = handler.getSupportedProtocols(); @@ -74,30 +90,17 @@ public final class SubProtocolHandlerRegistry { } continue; } - for (String protocol : protocols) { - SubProtocolHandler replaced = this.protocolHandlers.put(protocol, handler); - if (replaced != null) { - throw new IllegalStateException("Failed to map handler " + handler - + " to protocol '" + protocol + "', it is already mapped to handler " + replaced); - } - } + populateProtocolsForHandler(handler, protocols); } } + } - if (this.protocolHandlers.size() == 1 && defaultProtocolHandler == null) { - this.defaultProtocolHandler = this.protocolHandlers.values().iterator().next(); - } - else { - this.defaultProtocolHandler = defaultProtocolHandler; - if (this.protocolHandlers.isEmpty() && this.defaultProtocolHandler != null) { - List protocols = this.defaultProtocolHandler.getSupportedProtocols(); - for (String protocol : protocols) { - SubProtocolHandler replaced = this.protocolHandlers.put(protocol, this.defaultProtocolHandler); - if (replaced != null) { - throw new IllegalStateException("Failed to map handler " + this.defaultProtocolHandler - + " to protocol '" + protocol + "', it is already mapped to handler " + replaced); - } - } + private void populateProtocolsForHandler(SubProtocolHandler handler, List protocols) { + for (String protocol : protocols) { + SubProtocolHandler replaced = this.protocolHandlers.put(protocol, handler); + if (replaced != null) { + throw new IllegalStateException("Failed to map handler " + handler + + " to protocol '" + protocol + "', it is already mapped to handler " + replaced); } } }