diff --git a/build.gradle b/build.gradle index e11777cfcc..f68d3ad8c2 100644 --- a/build.gradle +++ b/build.gradle @@ -597,7 +597,8 @@ project('spring-integration-websocket') { compile project(":spring-integration-core") compile "org.springframework:spring-websocket:$springVersion" - testCompile "org.springframework:spring-webmvc:$springVersion" + compile ("org.springframework:spring-webmvc:$springVersion", optional) + testCompile("org.eclipse.jetty:jetty-webapp:$jettyVersion") { exclude group: "javax.servlet", module: "javax.servlet" } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index 9c8a863bd7..146c73cec4 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -2287,7 +2287,7 @@ @@ -2405,7 +2405,7 @@ diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java index 9011c41f87..07332d88f3 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/config/HttpIntegrationConfigurationInitializer.java @@ -40,7 +40,7 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi private static final Log logger = LogFactory.getLog(HttpIntegrationConfigurationInitializer.class); - private static boolean servletPresent = ClassUtils.isPresent("javax.servlet.Servlet", + private static final boolean servletPresent = ClassUtils.isPresent("javax.servlet.Servlet", HttpIntegrationConfigurationInitializer.class.getClassLoader()); @Override @@ -49,7 +49,8 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi this.registerRequestMappingHandlerMappingIfNecessary((BeanDefinitionRegistry) beanFactory); } else { - logger.warn("'IntegrationRequestMappingHandlerMapping' isn't registered because 'beanFactory' isn't an instance of `BeanDefinitionRegistry`."); + logger.warn("'IntegrationRequestMappingHandlerMapping' isn't registered because 'beanFactory'" + + " isn't an instance of `BeanDefinitionRegistry`."); } } @@ -64,12 +65,13 @@ public class HttpIntegrationConfigurationInitializer implements IntegrationConfi * the HTTP server components. */ private void registerRequestMappingHandlerMappingIfNecessary(BeanDefinitionRegistry registry) { - if (!registry.containsBeanDefinition(HttpContextUtils.HANDLER_MAPPING_BEAN_NAME) && servletPresent) { + if (servletPresent && !registry.containsBeanDefinition(HttpContextUtils.HANDLER_MAPPING_BEAN_NAME)) { BeanDefinitionBuilder requestMappingBuilder = BeanDefinitionBuilder.genericBeanDefinition(IntegrationRequestMappingHandlerMapping.class); requestMappingBuilder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); requestMappingBuilder.addPropertyValue(IntegrationNamespaceUtils.ORDER, 0); - registry.registerBeanDefinition(HttpContextUtils.HANDLER_MAPPING_BEAN_NAME, requestMappingBuilder.getBeanDefinition()); + registry.registerBeanDefinition(HttpContextUtils.HANDLER_MAPPING_BEAN_NAME, + requestMappingBuilder.getBeanDefinition()); } } 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 df1346186f..8144a599c4 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 @@ -16,6 +16,7 @@ package org.springframework.integration.websocket; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -23,6 +24,7 @@ import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; import org.springframework.http.HttpHeaders; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.web.socket.CloseStatus; @@ -66,8 +68,19 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine this.headers.setOrigin(origin); } + public void setHeadersMap(Map headers) { + Assert.notNull(headers); + HttpHeaders httpHeaders = new HttpHeaders(); + for (Map.Entry entry : headers.entrySet()) { + String[] values = StringUtils.commaDelimitedListToStringArray(entry.getValue()); + for (String v : values) { + httpHeaders.add(entry.getKey(), v); + } + } + setHeaders(httpHeaders); + } + public void setHeaders(HttpHeaders headers) { - this.headers.clear(); this.headers.putAll(headers); } @@ -121,8 +134,8 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine @Override public void start() { - this.connectionManager.start(); this.connectionLatch = new CountDownLatch(1); + this.connectionManager.start(); } @Override 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 new file mode 100644 index 0000000000..23a2d0f357 --- /dev/null +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java @@ -0,0 +1,189 @@ +package org.springframework.integration.websocket; + +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.web.socket.config.annotation.SockJsServiceRegistration; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistration; +import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; +import org.springframework.web.socket.server.HandshakeHandler; +import org.springframework.web.socket.server.HandshakeInterceptor; +import org.springframework.web.socket.sockjs.frame.SockJsMessageCodec; +import org.springframework.web.socket.sockjs.transport.TransportHandler; + +/** + * The {@link IntegrationWebSocketContainer} implementation for the {@code server} + * {@link org.springframework.web.socket.WebSocketHandler} registration. + *

+ * Registers an internal {@code IntegrationWebSocketContainer.IntegrationWebSocketHandler} + * for provided {@link #paths} with the {@link WebSocketHandlerRegistry}. + *

+ * The real registration is based on Spring Web-Socket infrastructure via {@link WebSocketConfigurer} + * implementation of this class. + * + * @author Artem Bilan + * @since 4.1 + */ +public class ServerWebSocketContainer extends IntegrationWebSocketContainer implements WebSocketConfigurer { + + private final String[] paths; + + private volatile HandshakeHandler handshakeHandler; + + private volatile HandshakeInterceptor[] interceptors; + + private SockJsServiceOptions sockJsServiceOptions; + + public ServerWebSocketContainer(String... paths) { + this.paths = paths; + } + + public ServerWebSocketContainer setHandshakeHandler(HandshakeHandler handshakeHandler) { + this.handshakeHandler = handshakeHandler; + return this; + } + + public ServerWebSocketContainer setInterceptors(HandshakeInterceptor[] interceptors) { + this.interceptors = interceptors; + return this; + } + + + public ServerWebSocketContainer withSockJs(SockJsServiceOptions... sockJsServiceOptions) { + if (ObjectUtils.isEmpty(sockJsServiceOptions)) { + this.sockJsServiceOptions = new SockJsServiceOptions(); + } + else { + Assert.state(sockJsServiceOptions.length == 1, "Only one 'sockJsServiceOptions' is applicable."); + this.sockJsServiceOptions = sockJsServiceOptions[0]; + } + return this; + } + + public void setSockJsServiceOptions(SockJsServiceOptions sockJsServiceOptions) { + this.sockJsServiceOptions = sockJsServiceOptions; + } + + @Override + public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { + WebSocketHandlerRegistration registration = registry.addHandler(this.webSocketHandler, this.paths) + .setHandshakeHandler(this.handshakeHandler) + .addInterceptors(this.interceptors); + 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); + } + } + + } + + /** + * @see org.springframework.web.socket.config.annotation.SockJsServiceRegistration + */ + public static class SockJsServiceOptions { + + private TaskScheduler taskScheduler; + + private String clientLibraryUrl; + + private Integer streamBytesLimit; + + private Boolean sessionCookieNeeded; + + private Long heartbeatTime; + + private Long disconnectDelay; + + private Integer httpMessageCacheSize; + + private Boolean webSocketEnabled; + + private TransportHandler[] transportHandlers; + + private SockJsMessageCodec messageCodec; + + public SockJsServiceOptions setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + return this; + } + + public SockJsServiceOptions setClientLibraryUrl(String clientLibraryUrl) { + this.clientLibraryUrl = clientLibraryUrl; + return this; + } + + public SockJsServiceOptions setStreamBytesLimit(int streamBytesLimit) { + this.streamBytesLimit = streamBytesLimit; + return this; + } + + public SockJsServiceOptions setSessionCookieNeeded(boolean sessionCookieNeeded) { + this.sessionCookieNeeded = sessionCookieNeeded; + return this; + } + + public SockJsServiceOptions setHeartbeatTime(long heartbeatTime) { + this.heartbeatTime = heartbeatTime; + return this; + } + + public SockJsServiceOptions setDisconnectDelay(long disconnectDelay) { + this.disconnectDelay = disconnectDelay; + return this; + } + + public SockJsServiceOptions setHttpMessageCacheSize(int httpMessageCacheSize) { + this.httpMessageCacheSize = httpMessageCacheSize; + return this; + } + + public SockJsServiceOptions setWebSocketEnabled(boolean webSocketEnabled) { + this.webSocketEnabled = webSocketEnabled; + return this; + } + + public SockJsServiceOptions setTransportHandlers(TransportHandler... transportHandlers) { + this.transportHandlers = transportHandlers; + return this; + } + + public SockJsServiceOptions setMessageCodec(SockJsMessageCodec messageCodec) { + this.messageCodec = messageCodec; + return this; + } + + } + +} diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/ClientWebSocketContainerParser.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/ClientWebSocketContainerParser.java new file mode 100644 index 0000000000..c42498ab37 --- /dev/null +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/ClientWebSocketContainerParser.java @@ -0,0 +1,72 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.websocket.config; + +import java.util.Map; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.websocket.ClientWebSocketContainer; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; + +/** + * The {@link AbstractSingleBeanDefinitionParser} implementation for + * the {@code } element. + * + * @author Artem Bilan + * @since 4.1 + */ +public class ClientWebSocketContainerParser extends AbstractSingleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return ClientWebSocketContainer.class; + } + + @Override + protected boolean shouldGenerateIdAsFallback() { + return true; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.addConstructorArgReference(element.getAttribute("client")) + .addConstructorArgValue(element.getAttribute("uri")); + String uriVariables = element.getAttribute("uri-variables"); + if (StringUtils.hasText(uriVariables)) { + builder.addConstructorArgValue(StringUtils.commaDelimitedListToStringArray(uriVariables)); + } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-buffer-size-limit"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-time-limit"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "origin"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); + + Element httpHeaders = DomUtils.getChildElementByTagName(element, "http-headers"); + if (httpHeaders != null) { + Map map = parserContext.getDelegate().parseMapElement(httpHeaders, builder.getBeanDefinition()); + builder.addPropertyValue("headersMap", map); + } + + } + +} diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/ServerWebSocketContainerParser.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/ServerWebSocketContainerParser.java new file mode 100644 index 0000000000..5363ad9637 --- /dev/null +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/ServerWebSocketContainerParser.java @@ -0,0 +1,100 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.websocket.config; + +import java.util.List; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanReference; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.ManagedList; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.websocket.ServerWebSocketContainer; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; + +/** + * The {@link org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser} implementation for + * the {@code } element. + * + * @author Artem Bilan + * @since 4.1 + */ +public class ServerWebSocketContainerParser extends AbstractSingleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return ServerWebSocketContainer.class; + } + + @Override + protected boolean shouldGenerateIdAsFallback() { + return true; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.addConstructorArgValue(element.getAttribute("path")); + + String handshakeInterceptors = element.getAttribute("handshake-interceptors"); + List handshakeInterceptorList = new ManagedList(); + String[] ids = StringUtils.commaDelimitedListToStringArray(handshakeInterceptors); + for (String id : ids) { + handshakeInterceptorList.add(new RuntimeBeanReference(id)); + } + builder.addPropertyValue("interceptors", handshakeInterceptorList); + + Element sockjs = DomUtils.getChildElementByTagName(element, "sockjs"); + + if (sockjs != null) { + BeanDefinitionBuilder sockjsBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ServerWebSocketContainer.SockJsServiceOptions.class); + IntegrationNamespaceUtils.setValueIfAttributeDefined(sockjsBuilder, sockjs, "client-library-url"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(sockjsBuilder, sockjs, "websocket-enabled", + "webSocketEnabled"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(sockjsBuilder, sockjs, "stream-bytes-limit"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(sockjsBuilder, sockjs, "session-cookie-needed"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(sockjsBuilder, sockjs, "heartbeat-time"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(sockjsBuilder, sockjs, "disconnect-delay"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(sockjsBuilder, sockjs, "message-cache-size", + "httpMessageCacheSize"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sockjsBuilder, sockjs, "scheduler", + "taskScheduler"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sockjsBuilder, sockjs, "message-codec"); + + String transportHandlers = sockjs.getAttribute("transport-handlers"); + if (StringUtils.hasText(transportHandlers)) { + List transportHandlerList = new ManagedList(); + ids = StringUtils.commaDelimitedListToStringArray(transportHandlers); + for (String id : ids) { + transportHandlerList.add(new RuntimeBeanReference(id)); + } + sockjsBuilder.addPropertyValue("transportHandlers", transportHandlerList); + } + builder.addPropertyValue("sockJsServiceOptions", sockjsBuilder.getBeanDefinition()); + } + + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "handshake-handler"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-buffer-size-limit"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-time-limit"); + } + +} diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketAdapterParsingUtils.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketAdapterParsingUtils.java new file mode 100644 index 0000000000..f2d5e12be6 --- /dev/null +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketAdapterParsingUtils.java @@ -0,0 +1,78 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.websocket.config; + +import java.util.List; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.config.BeanReference; +import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.ManagedList; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.websocket.support.SubProtocolHandlerRegistry; +import org.springframework.util.StringUtils; + +/** + * @author Artem Bilan + * @since 4.1 + */ +abstract class WebSocketAdapterParsingUtils { + + static void configureWebSocketAdapter(BeanDefinitionBuilder builder, ParserContext parserContext, Element element) { + String container = element.getAttribute("container"); + if (!StringUtils.hasText(container)) { + parserContext.getReaderContext().error("The 'container' is required", element); + } + builder.addConstructorArgReference(container); + + String protocolHandlers = element.getAttribute("protocol-handlers"); + boolean hasProtocolHandlers = StringUtils.hasText(protocolHandlers); + String defaultProtocolHandler = element.getAttribute("default-protocol-handler"); + boolean hasDefaultProtocolHandler = StringUtils.hasText(defaultProtocolHandler); + + if (hasProtocolHandlers || hasDefaultProtocolHandler) { + List protocolHandlerList = new ManagedList(); + String[] ids = StringUtils.commaDelimitedListToStringArray(protocolHandlers); + for (String id : ids) { + protocolHandlerList.add(new RuntimeBeanReference(id)); + } + BeanDefinitionBuilder protocolHandlerRegistryBuilder = + BeanDefinitionBuilder.genericBeanDefinition(SubProtocolHandlerRegistry.class) + .addConstructorArgValue(protocolHandlerList); + if (hasDefaultProtocolHandler) { + protocolHandlerRegistryBuilder.addConstructorArgReference(defaultProtocolHandler); + } + builder.addConstructorArgValue(protocolHandlerRegistryBuilder.getBeanDefinition()); + } + + String messageConverters = element.getAttribute("message-converters"); + if (StringUtils.hasText(messageConverters)) { + List messageConverterList = new ManagedList(); + String[] ids = StringUtils.commaDelimitedListToStringArray(messageConverters); + for (String id : ids) { + messageConverterList.add(new RuntimeBeanReference(id)); + } + builder.addPropertyValue("messageConverters", messageConverterList); + } + + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "merge-with-default-converters"); + } + +} diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketInboundChannelAdapterParser.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketInboundChannelAdapterParser.java new file mode 100644 index 0000000000..c984c657d2 --- /dev/null +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketInboundChannelAdapterParser.java @@ -0,0 +1,77 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.websocket.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.BeanDefinitionStoreException; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter; +import org.springframework.util.StringUtils; + +/** + * The {@link AbstractSingleBeanDefinitionParser} implementation for + * the {@code } element. + * + * @author Artem Bilan + * @since 4.1 + */ +public class WebSocketInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return WebSocketInboundChannelAdapter.class; + } + + @Override + protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) + throws BeanDefinitionStoreException { + String id = super.resolveId(element, definition, parserContext); + + if (!element.hasAttribute("channel")) { + // the created channel will get the 'id', so the adapter's bean name includes a suffix + id = id + ".adapter"; + } + if (!StringUtils.hasText(id)) { + id = BeanDefinitionReaderUtils.generateBeanName(definition, parserContext.getRegistry()); + } + + return id; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + WebSocketAdapterParsingUtils.configureWebSocketAdapter(builder, parserContext, element); + String channelName = element.getAttribute("channel"); + if (!StringUtils.hasText(channelName)) { + channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext); + } + builder.addPropertyReference("outputChannel", channelName); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "payload-type"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "use-broker"); + } + +} 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 new file mode 100644 index 0000000000..27646dfa8d --- /dev/null +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketIntegrationConfigurationInitializer.java @@ -0,0 +1,144 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.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.ListableBeanFactory; +import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.integration.config.IntegrationConfigurationInitializer; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.util.ClassUtils; +import org.springframework.web.servlet.HandlerMapping; +import org.springframework.web.servlet.handler.AbstractHandlerMapping; +import org.springframework.web.socket.config.annotation.DelegatingWebSocketConfiguration; +import org.springframework.web.socket.config.annotation.ServletWebSocketHandlerRegistry; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; + +/** + * The WebSocket Integration infrastructure {@code beanFactory} initializer. + * + * @author Artem Bilan + * @since 4.1 + */ +public class WebSocketIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer { + + private static final Log logger = LogFactory.getLog(WebSocketIntegrationConfigurationInitializer.class); + + private static final boolean servletPresent = ClassUtils.isPresent("javax.servlet.Servlet", + WebSocketIntegrationConfigurationInitializer.class.getClassLoader()); + + private static final String WEB_SOCKET_HANDLER_MAPPING_BEAN_NAME = "integrationWebSocketHandlerMapping"; + + @Override + public void initialize(ConfigurableListableBeanFactory beanFactory) throws BeansException { + if (beanFactory instanceof BeanDefinitionRegistry) { + this.registerEnableWebSocketIfNecessary((BeanDefinitionRegistry) beanFactory); + } + else { + logger.warn("'DelegatingWebSocketConfiguration' isn't registered because 'beanFactory'" + + " isn't an instance of `BeanDefinitionRegistry`."); + } + } + + /** + * Register a {@link WebSocketHandlerMappingFactoryBean} which could also be overridden + * by the user by simply using {@link org.springframework.web.socket.config.annotation.EnableWebSocket} + *

+ * In addition, checks if the {@code javax.servlet.Servlet} class is present on the classpath. + * When Spring Integration WebSocket support is used only as a WebSocket client, + * there is no reason to use and register the Spring WebSocket server components. + *

+ * Note, there is no XML equivalent for + * the {@link org.springframework.web.socket.config.annotation .EnableWebSocket} + * in the Spring WebSocket. therefore this registration can be used to process + * {@link WebSocketConfigurer} implementations without annotation configuration. + * From other side it can be used to replace + * {@link org.springframework.web.socket.config.annotation.EnableWebSocket} in the Spring Integration + * applications when {@link org.springframework.integration.config.EnableIntegration} is in use. + */ + private void registerEnableWebSocketIfNecessary(BeanDefinitionRegistry registry) { + if (servletPresent) { + if (!registry.containsBeanDefinition("defaultSockJsTaskScheduler")) { + BeanDefinitionBuilder sockJsTaskSchedulerBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ThreadPoolTaskScheduler.class) + .addPropertyValue("threadNamePrefix", "SockJS-") + .addPropertyValue("poolSize", Runtime.getRuntime().availableProcessors()) + .addPropertyValue("removeOnCancelPolicy", true); + + registry.registerBeanDefinition("defaultSockJsTaskScheduler", + sockJsTaskSchedulerBuilder.getBeanDefinition()); + } + + if (!registry.containsBeanDefinition(DelegatingWebSocketConfiguration.class.getName()) && + !registry.containsBeanDefinition(WEB_SOCKET_HANDLER_MAPPING_BEAN_NAME)) { + BeanDefinitionBuilder enableWebSocketBuilder = + BeanDefinitionBuilder.genericBeanDefinition(WebSocketHandlerMappingFactoryBean.class) + .setRole(BeanDefinition.ROLE_INFRASTRUCTURE) + .addConstructorArgReference("defaultSockJsTaskScheduler"); + + registry.registerBeanDefinition(WEB_SOCKET_HANDLER_MAPPING_BEAN_NAME, + enableWebSocketBuilder.getBeanDefinition()); + } + } + } + + private static class WebSocketHandlerMappingFactoryBean extends AbstractFactoryBean + implements ApplicationContextAware { + + private final ServletWebSocketHandlerRegistry registry; + + private ApplicationContext applicationContext; + + public WebSocketHandlerMappingFactoryBean(ThreadPoolTaskScheduler sockJsTaskScheduler) { + this.registry = new ServletWebSocketHandlerRegistry(sockJsTaskScheduler); + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + protected HandlerMapping createInstance() throws Exception { + Collection webSocketConfigurers = + ((ListableBeanFactory) getBeanFactory()).getBeansOfType(WebSocketConfigurer.class).values(); + for (WebSocketConfigurer configurer : webSocketConfigurers) { + configurer.registerWebSocketHandlers(this.registry); + } + AbstractHandlerMapping handlerMapping = this.registry.getHandlerMapping(); + handlerMapping.setApplicationContext(this.applicationContext); + return handlerMapping; + } + + @Override + public Class getObjectType() { + return HandlerMapping.class; + } + } + +} diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketNamespaceHandler.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketNamespaceHandler.java index a54ce94fb2..0804db075f 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketNamespaceHandler.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketNamespaceHandler.java @@ -25,7 +25,10 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa public class WebSocketNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { - + this.registerBeanDefinitionParser("server-container", new ServerWebSocketContainerParser()); + this.registerBeanDefinitionParser("client-container", new ClientWebSocketContainerParser()); + this.registerBeanDefinitionParser("inbound-channel-adapter", new WebSocketInboundChannelAdapterParser()); + this.registerBeanDefinitionParser("outbound-channel-adapter", new WebSocketOutboundMessageHandlerParser()); } } diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketOutboundMessageHandlerParser.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketOutboundMessageHandlerParser.java new file mode 100644 index 0000000000..3d51e1f119 --- /dev/null +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/config/WebSocketOutboundMessageHandlerParser.java @@ -0,0 +1,43 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.websocket.config; + +import org.w3c.dom.Element; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; +import org.springframework.integration.websocket.outbound.WebSocketOutboundMessageHandler; + +/** + * The {@link AbstractOutboundChannelAdapterParser} implementation for + * the {@code } element. + * + * @author Artem Bilan + * @since 4.1 + */ +public class WebSocketOutboundMessageHandlerParser extends AbstractOutboundChannelAdapterParser { + + @Override + protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(WebSocketOutboundMessageHandler.class); + WebSocketAdapterParsingUtils.configureWebSocketAdapter(builder, parserContext, element); + return builder.getBeanDefinition(); + } + +} 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 627001a8e6..c0413c8b26 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 @@ -17,16 +17,18 @@ package org.springframework.integration.websocket.inbound; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.ListIterator; +import java.util.Map; import java.util.concurrent.atomic.AtomicReference; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.endpoint.MessageProducerSupport; -import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.json.JacksonJsonUtils; import org.springframework.integration.websocket.IntegrationWebSocketContainer; +import org.springframework.integration.websocket.ServerWebSocketContainer; import org.springframework.integration.websocket.WebSocketListener; import org.springframework.integration.websocket.support.PassThruSubProtocolHandler; import org.springframework.integration.websocket.support.SubProtocolHandlerRegistry; @@ -42,6 +44,9 @@ import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.converter.StringMessageConverter; import org.springframework.messaging.simp.SimpMessageHeaderAccessor; import org.springframework.messaging.simp.SimpMessageType; +import org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler; +import org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler; +import org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.MimeTypeUtils; @@ -73,7 +78,9 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple private final IntegrationWebSocketContainer webSocketContainer; - private final SubProtocolHandlerRegistry protocolHandlerContainer; + private final boolean server; + + private final SubProtocolHandlerRegistry subProtocolHandlerRegistry; private final MessageChannel subProtocolHandlerChannel; @@ -85,6 +92,10 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple private volatile boolean active; + private volatile boolean useBroker; + + private AbstractBrokerMessageHandler brokerHandler; + public WebSocketInboundChannelAdapter(IntegrationWebSocketContainer webSocketContainer) { this(webSocketContainer, new SubProtocolHandlerRegistry(new PassThruSubProtocolHandler())); } @@ -94,26 +105,13 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple Assert.notNull(webSocketContainer, "'webSocketContainer' must not be null"); Assert.notNull(protocolHandlerRegistry, "'protocolHandlerRegistry' must not be null"); this.webSocketContainer = webSocketContainer; - this.protocolHandlerContainer = protocolHandlerRegistry; + this.server = this.webSocketContainer instanceof ServerWebSocketContainer; + this.subProtocolHandlerRegistry = protocolHandlerRegistry; this.subProtocolHandlerChannel = new FixedSubscriberChannel(new MessageHandler() { @Override public void handleMessage(Message message) throws MessagingException { - Object payload = WebSocketInboundChannelAdapter.this.messageConverter.fromMessage(message, - WebSocketInboundChannelAdapter.this.payloadType.get()); - SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message); - SimpMessageType messageType = headerAccessor.getMessageType(); - if (messageType == null || SimpMessageType.MESSAGE.equals(messageType)) { - headerAccessor.removeHeader(SimpMessageHeaderAccessor.NATIVE_HEADERS); - sendMessage(MessageBuilder.withPayload(payload).copyHeaders(headerAccessor.toMap()).build()); - } - else { - if (logger.isDebugEnabled()) { - logger.debug("Messages with non 'SimpMessageType.MESSAGE' type are ignored for sending to the " + - "'outputChannel'. They have to be emitted as 'ApplicationEvent's " + - "from the 'SubProtocolHandler'. Received message: " + message); - } - } + handleMessageAndSend(message); } }); @@ -149,6 +147,20 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple this.payloadType.set(payloadType); } + /** + * Specify if this adapter should use an existing single {@link AbstractBrokerMessageHandler} + * bean for {@code non-MESSAGE} {@link org.springframework.web.socket.WebSocketMessage}s + * and to route messages with broker destinations. + * Since only single {@link AbstractBrokerMessageHandler} bean is allowed in the current + * application context, the algorithm to lookup the former by type, rather than applying + * the bean reference. + * This is used only on server side and is ignored from client side. + * @param useBroker the boolean flag. + */ + public void setUseBroker(boolean useBroker) { + this.useBroker = useBroker; + } + @Override protected void onInit() { super.onInit(); @@ -156,7 +168,9 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple if (!CollectionUtils.isEmpty(this.messageConverters)) { List converters = this.messageConverter.getConverters(); if (this.mergeWithDefaultConverters) { - for (ListIterator iterator = this.messageConverters.listIterator(); iterator.hasPrevious(); ) { + ListIterator iterator = + this.messageConverters.listIterator(this.messageConverters.size()); + while (iterator.hasPrevious()) { MessageConverter converter = iterator.previous(); converters.add(0, converter); } @@ -166,17 +180,31 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple converters.addAll(this.messageConverters); } } + if (this.server && this.useBroker) { + Map brokers = getApplicationContext() + .getBeansOfType(AbstractBrokerMessageHandler.class); + for (AbstractBrokerMessageHandler broker : brokers.values()) { + if (broker instanceof SimpleBrokerMessageHandler || broker instanceof StompBrokerRelayMessageHandler) { + this.brokerHandler = broker; + break; + } + } + if (this.brokerHandler == null) { + logger.warn("'AbstractBrokerMessageHandler' isn't present in the application context. " + + "The non-MESSAGE WebSocketMessages will be ignored."); + } + } } @Override public List getSubProtocols() { - return this.protocolHandlerContainer.getSubProtocols(); + return this.subProtocolHandlerRegistry.getSubProtocols(); } @Override public void afterSessionStarted(WebSocketSession session) throws Exception { if (isActive()) { - this.protocolHandlerContainer.findProtocolHandler(session) + this.subProtocolHandlerRegistry.findProtocolHandler(session) .afterSessionStarted(session, this.subProtocolHandlerChannel); } } @@ -184,7 +212,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple @Override public void afterSessionEnded(WebSocketSession session, CloseStatus closeStatus) throws Exception { if (isActive()) { - this.protocolHandlerContainer.findProtocolHandler(session) + this.subProtocolHandlerRegistry.findProtocolHandler(session) .afterSessionEnded(session, closeStatus, this.subProtocolHandlerChannel); } } @@ -192,7 +220,7 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple @Override public void onMessage(WebSocketSession session, WebSocketMessage webSocketMessage) throws Exception { if (isActive()) { - this.protocolHandlerContainer.findProtocolHandler(session) + this.subProtocolHandlerRegistry.findProtocolHandler(session) .handleMessageFromClient(session, webSocketMessage, this.subProtocolHandlerChannel); } } @@ -222,4 +250,42 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport imple return this.active; } + private void handleMessageAndSend(Message message) { + Object payload = this.messageConverter.fromMessage(message, + this.payloadType.get()); + SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(message); + SimpMessageType messageType = headerAccessor.getMessageType(); + if ((messageType == null || SimpMessageType.MESSAGE.equals(messageType)) + && !checkDestinationPrefix(headerAccessor.getDestination())) { + headerAccessor.removeHeader(SimpMessageHeaderAccessor.NATIVE_HEADERS); + sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headerAccessor.toMap()).build()); + } + else { + if (this.brokerHandler != null) { + this.brokerHandler.handleMessage(message); + } + else if (logger.isDebugEnabled()) { + logger.debug("Messages with non 'SimpMessageType.MESSAGE' type are ignored for sending to the " + + "'outputChannel'. They have to be emitted as 'ApplicationEvent's " + + "from the 'SubProtocolHandler'. Or using 'AbstractBrokerMessageHandler'(useBroker = true) " + + "from server side. Received message: " + message); + } + } + } + + private boolean checkDestinationPrefix(String destination) { + if (this.brokerHandler != null) { + Collection destinationPrefixes = this.brokerHandler.getDestinationPrefixes(); + if ((destination == null) || CollectionUtils.isEmpty(destinationPrefixes)) { + return false; + } + for (String prefix : destinationPrefixes) { + if (destination.startsWith(prefix)) { + return true; + } + } + } + return false; + } + } 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 10df329fc2..2751eca4ed 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 @@ -65,7 +65,7 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler { private final IntegrationWebSocketContainer webSocketContainer; - private final SubProtocolHandlerRegistry protocolHandlerContainer; + private final SubProtocolHandlerRegistry subProtocolHandlerRegistry; private final boolean client; @@ -83,7 +83,7 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler { Assert.notNull(protocolHandlerRegistry, "'protocolHandlerRegistry' must not be null"); this.webSocketContainer = webSocketContainer; this.client = webSocketContainer instanceof ClientWebSocketContainer; - this.protocolHandlerContainer = protocolHandlerRegistry; + this.subProtocolHandlerRegistry = protocolHandlerRegistry; List subProtocols = protocolHandlerRegistry.getSubProtocols(); this.webSocketContainer.addSupportedProtocols(subProtocols.toArray(new String[subProtocols.size()])); } @@ -119,7 +119,9 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler { if (!CollectionUtils.isEmpty(this.messageConverters)) { List converters = this.messageConverter.getConverters(); if (this.mergeWithDefaultConverters) { - for (ListIterator iterator = this.messageConverters.listIterator(); iterator.hasPrevious(); ) { + ListIterator iterator = + this.messageConverters.listIterator(this.messageConverters.size()); + while (iterator.hasPrevious()) { MessageConverter converter = iterator.previous(); converters.add(0, converter); } @@ -135,7 +137,7 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler { protected void handleMessageInternal(Message message) throws Exception { String sessionId = null; if (!this.client) { - sessionId = this.protocolHandlerContainer.resolveSessionId(message); + sessionId = this.subProtocolHandlerRegistry.resolveSessionId(message); if (sessionId == null) { throw new IllegalArgumentException("The WebSocket 'sessionId' is required in the MessageHeaders"); } @@ -146,7 +148,7 @@ public class WebSocketOutboundMessageHandler extends AbstractMessageHandler { headers.setLeaveMutable(true); headers.setMessageTypeIfNotSet(SimpMessageType.MESSAGE); Message messageToSend = this.messageConverter.toMessage(message.getPayload(), headers.getMessageHeaders()); - this.protocolHandlerContainer.findProtocolHandler(session).handleMessageToClient(session, messageToSend); + this.subProtocolHandlerRegistry.findProtocolHandler(session).handleMessageToClient(session, messageToSend); } catch (SessionLimitExceededException ex) { try { 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 aa265ea0d0..02aee9f855 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 @@ -85,7 +85,7 @@ public final class SubProtocolHandlerRegistry { } else { this.defaultProtocolHandler = defaultProtocolHandler; - if (this.protocolHandlers.isEmpty()) { + if (this.protocolHandlers.isEmpty() && this.defaultProtocolHandler != null) { List protocols = this.defaultProtocolHandler.getSupportedProtocols(); for (String protocol : protocols) { SubProtocolHandler replaced = this.protocolHandlers.put(protocol, this.defaultProtocolHandler); diff --git a/spring-integration-websocket/src/main/resources/META-INF/spring.factories b/spring-integration-websocket/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000000..01c50fa257 --- /dev/null +++ b/spring-integration-websocket/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.integration.config.IntegrationConfigurationInitializer=\ +org.springframework.integration.websocket.config.WebSocketIntegrationConfigurationInitializer diff --git a/spring-integration-websocket/src/main/resources/org/springframework/integration/websocket/config/spring-integration-websocket-4.1.xsd b/spring-integration-websocket/src/main/resources/org/springframework/integration/websocket/config/spring-integration-websocket-4.1.xsd index 438a1912c1..ff79e5954c 100644 --- a/spring-integration-websocket/src/main/resources/org/springframework/integration/websocket/config/spring-integration-websocket-4.1.xsd +++ b/spring-integration-websocket/src/main/resources/org/springframework/integration/websocket/config/spring-integration-websocket-4.1.xsd @@ -1,17 +1,17 @@ + xmlns:xsd="http://www.w3.org/2001/XMLSchema" + xmlns:beans="http://www.springframework.org/schema/beans" + xmlns:tool="http://www.springframework.org/schema/tool" + xmlns:integration="http://www.springframework.org/schema/integration" + targetNamespace="http://www.springframework.org/schema/integration/websocket" + elementFormDefault="qualified" + attributeFormDefault="unqualified"> + schemaLocation="http://www.springframework.org/schema/integration/spring-integration-4.1.xsd"/> + + + + Configures the 'ClientWebSocketContainer' bean. + + + + + + + + + + + + + + + + + + + + The reference to a 'WebSocketClient' bean, which encapsulates the low-level + connection and WebSocketSession handling operations. Required. + + + + + + + + + + + + The 'uri' or 'uriTemplate' to the target WebSocket service. If is used as 'uriTemplate' + with URI variable placeholders the 'uri-variables' attribute is required. + + + + + + + Comma-separated values for the URI variable placeholders within 'uri'. + See 'UriComponents.expand(Object... uriVariableValues)'. + + + + + + + The 'Origin' HTTP header value. + + + + + + + The WebSocket session 'send' timeout limit. + + + + + + + + + + The WebSocket session 'send' message size limit. + + + + + + + + + + + + + + Configures the 'ServerWebSocketContainer' bean. + + + + + + + + + + Transports with no native cross-domain communication (e.g. "eventsource", + "htmlfile") must get a simple page from the "foreign" domain in an invisible + iframe so that code in the iframe can run from a domain local to the SockJS + server. Since the iframe needs to load the SockJS javascript client library, + this property allows specifying where to load it from. + + By default this is set to point to + "https://d1fxtkz8shb9d2.cloudfront.net/sockjs-0.3.4.min.js". However it can + also be set to point to a URL served by the application. + + Note that it's possible to specify a relative URL in which case the URL + must be relative to the iframe URL. For example assuming a SockJS endpoint + mapped to "/sockjs", and resulting iframe URL "/sockjs/iframe.html", then the + The relative URL must start with "../../" to traverse up to the location + above the SockJS mapping. In case of a prefix-based Servlet mapping one more + traversal may be needed. + + + + + + + Minimum number of bytes that can be send over a single HTTP streaming request before + it will be closed. + Defaults to 128K (i.e. 128 1024). + + + + + + + + + + The "cookie_needed" value in the response from the SockJs "/info" endpoint. + This property indicates whether the use of a JSESSIONID cookie is required for the + application to function correctly, + e.g. for load balancing or in Java Servlet containers for the use of an HTTP + session. + + + + + + + + + + The amount of time in milliseconds when the server has not sent any messages and + after which the server + should send a heartbeat frame to the client in order to keep the connection from + breaking. + The default value is 25,000 (25 seconds). + + + + + + + + + + The amount of time in milliseconds before a client is considered disconnected after + not having + a receiving connection, i.e. an active connection over which the server can send + data to the client. + The default value is 5000. + + + + + + + + + + The number of server-to-client messages that a session can cache while waiting for + the next HTTP polling request from the client. + The default size is 100. + + + + + + + + + + Some load balancers don't support websockets. Set this option to "false" to disable + the WebSocket transport on the server side. + The default value is "true". + + + + + + + + + + The bean name of a TaskScheduler; a new ThreadPoolTaskScheduler instance will be + created if no value is provided. + This scheduler instance will be used for scheduling heart-beat messages. + + + + + + + + + + + + The bean name of a SockJsMessageCodec to use for encoding and decoding SockJS + messages. + By default Jackson2SockJsMessageCodec is used requiring the Jackson library to be + present on the classpath. + + + + + + + + + + + + List of 'org.springframework.web.socket.sockjs.transport.TransportHandler' + bean references. + + + + + + + + + + + A path (or comma-separated paths) that maps a particular request to a 'WebSocketHandler'. + Exact path mapping URIs (such as "/myPath") are supported as well + as ant-style path patterns (such as /myPath/**). + + + + + + + The reference to a 'HandshakeHandler' bean. + + + + + + + + + + + + List of 'org.springframework.web.socket.server.HandshakeInterceptor' bean references. + + + + + + + The WebSocket session 'send' timeout limit. + + + + + + + + + + The WebSocket session 'send' message size limit. + + + + + + + + + + + + + Configures an endpoint that will send a WebSocket Message to the provided + 'IntegrationWebSocketContainer'. + + + + + + + + + + + + + + + + + + Configures an endpoint that will receive WebSocket Messages from the provided + 'IntegrationWebSocketContainer' forward converted messages to a Message Channel. + + + + + + + + + Message Channel to which error Messages should be sent. + + + + + + + + + + + + Maximum amount of time in milliseconds to wait when sending a message + to the channel if such channel may block. + For example, a Queue Channel can block until space is available + if its maximum capacity has been reached. + + + + + + + Fully qualified name of the java type for the target `payload` + to convert from the incoming WebSocketMessage. + + + + + + + Flag to indicate if this adapter will send non-MESSAGE type WebSocketMessages + and messages with broker destinations to the `AbstractBrokerMessageHandler` + from the application context. + If the `AbstractBrokerMessageHandler` bean isn't present the warn log is emitted + and adapter behaviour is falling back to default like this attribute is 'false'. + This attribute is used only on server side. On client side it is ignored. + Defaults to 'false'. + + + + + + + + + + + + + + + Base type for the 'inbound-channel-adapter' and 'outbound-channel-adapter' elements. + + + + + + The reference to the 'IntegrationWebSocketContainer' bean, which encapsulates the low-level + connection and WebSocketSession handling operations. Required. + + + + + + + + + + + + Optional reference to a 'org.springframework.web.socket.messaging.SubProtocolHandler' instance. + It is used when the client did not request a sub-protocol or it is a single protocol-handler. + If this reference or 'protocol-handlers' list aren't provided the `PassThruSubProtocolHandler` + is used by default. + + + + + + + + + + + + List of 'org.springframework.web.socket.messaging.SubProtocolHandler' bean references + for this Channel Adapter. If only single bean reference is provided and 'default-protocol-handler' + isn't provided, that single 'SubProtocolHandler' will be presented as 'default-protocol-handler'. + If this attribute or 'default-protocol-handler' aren't provided the `PassThruSubProtocolHandler` + is used by default. + + + + + + + List of 'org.springframework.messaging.converter.MessageConverter' bean references + for this Channel Adapter. + + + + + + + Flag to indicate if the default converters should be registered after any custom + converters. This flag is used only if message-converters + are provided, otherwise all default converters will be registered. + Defaults to 'false'. + + + + + + + + + diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java index 08ed4037c0..4ce3e296c0 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java @@ -96,6 +96,7 @@ public class ClientWebSocketContainerTests { assertThat(messageListener.message, instanceOf(PongMessage.class)); } + private class TestWebSocketListener implements WebSocketListener { public boolean started; @@ -126,6 +127,7 @@ public class ClientWebSocketContainerTests { public List getSubProtocols() { return Collections.singletonList("v10.stomp"); } + } } diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/JettyWebSocketTestServer.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/JettyWebSocketTestServer.java index 5e3e2c684e..d770556af8 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/JettyWebSocketTestServer.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/JettyWebSocketTestServer.java @@ -52,7 +52,7 @@ public class JettyWebSocketTestServer implements InitializingBean, DisposableBea } public AnnotationConfigWebApplicationContext getServerContext() { - return serverContext; + return this.serverContext; } public String getWsBaseUrl() { diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TestServerConfig.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TestServerConfig.java index 7d5cba95dd..d432264ea1 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TestServerConfig.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/TestServerConfig.java @@ -21,6 +21,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.integration.channel.AbstractSubscribableChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.simp.SimpMessageHeaderAccessor; @@ -40,6 +41,11 @@ import org.springframework.web.socket.messaging.SubProtocolWebSocketHandler; */ @Configuration @EnableWebSocket +/* + * According to the WebSocketIntegrationConfigurationInitializer with usage of @EnableIntegration there is no need to + * use @EnableWebSocket anymore. They are left here both to check consistency of registration algorithm. + */ +@EnableIntegration public class TestServerConfig implements WebSocketConfigurer { @Bean diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java index a3441d42f3..eda93c8c71 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java @@ -132,7 +132,7 @@ public class StompIntegrationTests { this.webSocketOutputChannel.send(message); this.webSocketOutputChannel.send(message2); - Message receive = webSocketInputChannel.receive(1000); + Message receive = webSocketInputChannel.receive(10000); assertNotNull(receive); assertEquals("6", receive.getPayload()); } @@ -155,7 +155,7 @@ public class StompIntegrationTests { this.webSocketOutputChannel.send(message); this.webSocketOutputChannel.send(message2); - Message receive = webSocketInputChannel.receive(1000); + Message receive = webSocketInputChannel.receive(10000); assertNotNull(receive); assertEquals("10", receive.getPayload()); } @@ -236,7 +236,7 @@ public class StompIntegrationTests { this.webSocketOutputChannel.send(message); this.webSocketOutputChannel.send(message2); - Message receive = webSocketInputChannel.receive(5000); + Message receive = webSocketInputChannel.receive(10000); assertNotNull(receive); assertEquals("Hello Bob", receive.getPayload()); } diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/WebSocketClientTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/WebSocketClientTests.java index 74dd8b08ec..11ce949a1a 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/WebSocketClientTests.java +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/WebSocketClientTests.java @@ -28,8 +28,6 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.Poller; @@ -74,9 +72,6 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport; @DirtiesContext public class WebSocketClientTests { - @Value("#{server.serverContext}") - private ApplicationContext serverContext; - @Autowired @Qualifier("webSocketOutputChannel") private MessageChannel webSocketOutputChannel; diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/config/WebSocketParserTests-context.xml b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/config/WebSocketParserTests-context.xml new file mode 100644 index 0000000000..c4240a4557 --- /dev/null +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/config/WebSocketParserTests-context.xml @@ -0,0 +1,104 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/config/WebSocketParserTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/config/WebSocketParserTests.java new file mode 100644 index 0000000000..2e3f9af0c5 --- /dev/null +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/config/WebSocketParserTests.java @@ -0,0 +1,284 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.websocket.config; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.integration.support.converter.MapMessageConverter; +import org.springframework.integration.support.converter.SimpleMessageConverter; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.websocket.IntegrationWebSocketContainer; +import org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter; +import org.springframework.integration.websocket.outbound.WebSocketOutboundMessageHandler; +import org.springframework.integration.websocket.support.PassThruSubProtocolHandler; +import org.springframework.integration.websocket.support.SubProtocolHandlerRegistry; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.converter.CompositeMessageConverter; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.converter.StringMessageConverter; +import org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.web.servlet.HandlerMapping; +import org.springframework.web.socket.WebSocketHttpHeaders; +import org.springframework.web.socket.client.WebSocketClient; +import org.springframework.web.socket.messaging.StompSubProtocolHandler; +import org.springframework.web.socket.server.HandshakeHandler; +import org.springframework.web.socket.server.HandshakeInterceptor; +import org.springframework.web.socket.sockjs.frame.SockJsMessageCodec; +import org.springframework.web.socket.sockjs.transport.TransportHandler; +import org.springframework.web.socket.sockjs.transport.TransportHandlingSockJsService; +import org.springframework.web.socket.sockjs.transport.TransportType; + +/** + * @author Artem Bilan + * @since 4.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class WebSocketParserTests { + + @Autowired + @Qualifier("integrationWebSocketHandlerMapping") + private HandlerMapping handlerMapping; + + @Autowired + @Qualifier("serverWebSocketContainer") + private IntegrationWebSocketContainer serverWebSocketContainer; + + @Autowired + private TaskScheduler taskScheduler; + + @Autowired + private HandshakeHandler handshakeHandler; + + @Autowired + private HandshakeInterceptor handshakeInterceptor; + + @Autowired + private SockJsMessageCodec sockJsMessageCodec; + + @Autowired + @Qualifier("defaultInboundAdapter.adapter") + private WebSocketInboundChannelAdapter defaultInboundAdapter; + + @Autowired + private AbstractBrokerMessageHandler brokerHandler; + + @Autowired + @Qualifier("clientWebSocketContainer") + private IntegrationWebSocketContainer clientWebSocketContainer; + + @Autowired + @Qualifier("customInboundAdapter") + private WebSocketInboundChannelAdapter customInboundAdapter; + + @Autowired + private MessageChannel clientInboundChannel; + + @Autowired + private MessageChannel errorChannel; + + @Autowired + private StompSubProtocolHandler stompSubProtocolHandler; + + @Autowired + private SimpleMessageConverter simpleMessageConverter; + + @Autowired + private MapMessageConverter mapMessageConverter; + + @Autowired + private WebSocketClient webSocketClient; + + @Autowired + @Qualifier("defaultOutboundAdapter.handler") + private WebSocketOutboundMessageHandler defaultOutboundAdapter; + + @Autowired + @Qualifier("customOutboundAdapter.handler") + private WebSocketOutboundMessageHandler customOutboundAdapter; + + @Test + public void testDefaultInboundChannelAdapterAndServerContainer() { + Map urlMap = TestUtils.getPropertyValue(this.handlerMapping, "urlMap", Map.class); + assertEquals(1, urlMap.size()); + assertTrue(urlMap.containsKey("/ws/**")); + Object mappedHandler = urlMap.get("/ws/**"); + //WebSocketHttpRequestHandler -> ExceptionWebSocketHandlerDecorator - > LoggingWebSocketHandlerDecorator + // -> IntegrationWebSocketContainer$IntegrationWebSocketHandler + assertSame(TestUtils.getPropertyValue(this.serverWebSocketContainer, "webSocketHandler"), + TestUtils.getPropertyValue(mappedHandler, "webSocketHandler.delegate.delegate")); + assertSame(this.handshakeHandler, + TestUtils.getPropertyValue(this.serverWebSocketContainer, "handshakeHandler")); + HandshakeInterceptor[] interceptors = + TestUtils.getPropertyValue(this.serverWebSocketContainer, "interceptors", HandshakeInterceptor[].class); + assertEquals(1, interceptors.length); + assertSame(this.handshakeInterceptor, interceptors[0]); + assertEquals(100, TestUtils.getPropertyValue(this.serverWebSocketContainer, "sendTimeLimit")); + assertEquals(100000, TestUtils.getPropertyValue(this.serverWebSocketContainer, "sendBufferSizeLimit")); + + TransportHandlingSockJsService sockJsService = + TestUtils.getPropertyValue(mappedHandler, "sockJsService", TransportHandlingSockJsService.class); + assertSame(this.taskScheduler, sockJsService.getTaskScheduler()); + assertSame(this.sockJsMessageCodec, sockJsService.getMessageCodec()); + Map transportHandlers = sockJsService.getTransportHandlers(); + + //If "handshake-handler" is provided, "transport-handlers" isn't allowed + assertEquals(8, transportHandlers.size()); + assertSame(this.handshakeHandler, + TestUtils.getPropertyValue(transportHandlers.get(TransportType.WEBSOCKET), "handshakeHandler")); + assertEquals(4000L, sockJsService.getDisconnectDelay()); + assertEquals(30000L, sockJsService.getHeartbeatTime()); + assertEquals(10000, sockJsService.getHttpMessageCacheSize()); + assertEquals(2000, sockJsService.getStreamBytesLimit()); + assertEquals("https://foo.sock.js", sockJsService.getSockJsClientLibraryUrl()); + assertFalse(sockJsService.isSessionCookieNeeded()); + assertFalse(sockJsService.isWebSocketEnabled()); + + assertSame(this.serverWebSocketContainer, + TestUtils.getPropertyValue(this.defaultInboundAdapter, "webSocketContainer")); + assertNull(TestUtils.getPropertyValue(this.defaultInboundAdapter, "messageConverters")); + assertEquals(TestUtils.getPropertyValue(this.defaultInboundAdapter, "messageConverter.converters"), + TestUtils.getPropertyValue(this.defaultInboundAdapter, "defaultConverters")); + assertEquals(String.class, + TestUtils.getPropertyValue(this.defaultInboundAdapter, "payloadType", AtomicReference.class).get()); + assertTrue(TestUtils.getPropertyValue(this.defaultInboundAdapter, "useBroker", Boolean.class)); + assertSame(this.brokerHandler, TestUtils.getPropertyValue(this.defaultInboundAdapter, "brokerHandler")); + + SubProtocolHandlerRegistry subProtocolHandlerRegistry = TestUtils.getPropertyValue(this.defaultInboundAdapter, + "subProtocolHandlerRegistry", SubProtocolHandlerRegistry.class); + assertThat(TestUtils.getPropertyValue(subProtocolHandlerRegistry, "defaultProtocolHandler"), + instanceOf(PassThruSubProtocolHandler.class)); + assertTrue(TestUtils.getPropertyValue(subProtocolHandlerRegistry, "protocolHandlers", Map.class).isEmpty()); + } + + @Test + public void testCustomInboundChannelAdapterAndClientContainer() throws URISyntaxException { + assertSame(this.clientInboundChannel, TestUtils.getPropertyValue(this.customInboundAdapter, "outputChannel")); + assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customInboundAdapter, "errorChannel")); + assertSame(this.clientWebSocketContainer, + TestUtils.getPropertyValue(this.customInboundAdapter, "webSocketContainer")); + assertEquals(2000L, TestUtils.getPropertyValue(this.customInboundAdapter, "messagingTemplate.sendTimeout")); + assertEquals(200, TestUtils.getPropertyValue(this.customInboundAdapter, "phase")); + assertFalse(TestUtils.getPropertyValue(this.customInboundAdapter, "autoStartup", Boolean.class)); + assertEquals(Integer.class, + TestUtils.getPropertyValue(this.customInboundAdapter, "payloadType", AtomicReference.class).get()); + SubProtocolHandlerRegistry subProtocolHandlerRegistry = TestUtils.getPropertyValue(this.customInboundAdapter, + "subProtocolHandlerRegistry", SubProtocolHandlerRegistry.class); + assertSame(this.stompSubProtocolHandler, TestUtils.getPropertyValue(subProtocolHandlerRegistry, + "defaultProtocolHandler")); + Map protocolHandlers = + TestUtils.getPropertyValue(subProtocolHandlerRegistry, "protocolHandlers", Map.class); + assertEquals(3, protocolHandlers.size()); + //PassThruSubProtocolHandler is ignored because it doesn't provide any 'protocol' by default. + //See warn log message. + for (Object handler : protocolHandlers.values()) { + assertSame(this.stompSubProtocolHandler, handler); + } + + assertTrue(TestUtils.getPropertyValue(this.customInboundAdapter, "mergeWithDefaultConverters", Boolean.class)); + CompositeMessageConverter compositeMessageConverter = TestUtils.getPropertyValue(this.customInboundAdapter, + "messageConverter", CompositeMessageConverter.class); + List converters = compositeMessageConverter.getConverters(); + assertEquals(5, converters.size()); + assertSame(this.simpleMessageConverter, converters.get(0)); + assertSame(this.mapMessageConverter, converters.get(1)); + assertThat(converters.get(2), instanceOf(StringMessageConverter.class)); + + //Test ClientWebSocketContainer parser + assertSame(this.customInboundAdapter, + TestUtils.getPropertyValue(this.clientWebSocketContainer, "messageListener")); + assertEquals(100, TestUtils.getPropertyValue(this.clientWebSocketContainer, "sendTimeLimit")); + assertEquals(1000, TestUtils.getPropertyValue(this.clientWebSocketContainer, "sendBufferSizeLimit")); + assertEquals(new URI("ws://foo.bar/ws?service=user"), + TestUtils.getPropertyValue(this.clientWebSocketContainer, "connectionManager.uri", URI.class)); + assertSame(this.webSocketClient, + TestUtils.getPropertyValue(this.clientWebSocketContainer, "connectionManager.client")); + assertEquals(100, TestUtils.getPropertyValue(this.clientWebSocketContainer, "connectionManager.phase")); + WebSocketHttpHeaders headers = TestUtils.getPropertyValue(this.clientWebSocketContainer, "headers", + WebSocketHttpHeaders.class); + assertEquals("FOO", headers.getOrigin()); + assertEquals(Arrays.asList("BAR", "baz"), headers.get("FOO")); + } + + @Test + public void testDefaultOutboundChannelAdapter() { + assertSame(this.serverWebSocketContainer, + TestUtils.getPropertyValue(this.defaultOutboundAdapter, "webSocketContainer")); + assertNull(TestUtils.getPropertyValue(this.defaultOutboundAdapter, "messageConverters")); + assertEquals(TestUtils.getPropertyValue(this.defaultOutboundAdapter, "messageConverter.converters"), + TestUtils.getPropertyValue(this.defaultOutboundAdapter, "defaultConverters")); + SubProtocolHandlerRegistry subProtocolHandlerRegistry = TestUtils.getPropertyValue(this.defaultOutboundAdapter, + "subProtocolHandlerRegistry", SubProtocolHandlerRegistry.class); + assertThat(TestUtils.getPropertyValue(subProtocolHandlerRegistry, "defaultProtocolHandler"), + instanceOf(PassThruSubProtocolHandler.class)); + assertTrue(TestUtils.getPropertyValue(subProtocolHandlerRegistry, "protocolHandlers", Map.class).isEmpty()); + assertFalse(TestUtils.getPropertyValue(this.defaultOutboundAdapter, "client", Boolean.class)); + } + + @Test + public void testCustomOutboundChannelAdapter() throws URISyntaxException { + assertSame(this.clientWebSocketContainer, + TestUtils.getPropertyValue(this.customOutboundAdapter, "webSocketContainer")); + + SubProtocolHandlerRegistry subProtocolHandlerRegistry = TestUtils.getPropertyValue(this.customOutboundAdapter, + "subProtocolHandlerRegistry", SubProtocolHandlerRegistry.class); + assertSame(this.stompSubProtocolHandler, TestUtils.getPropertyValue(subProtocolHandlerRegistry, + "defaultProtocolHandler")); + Map protocolHandlers = + TestUtils.getPropertyValue(subProtocolHandlerRegistry, "protocolHandlers", Map.class); + assertEquals(3, protocolHandlers.size()); + //PassThruSubProtocolHandler is ignored because it doesn't provide any 'protocol' by default. + //See warn log message. + for (Object handler : protocolHandlers.values()) { + assertSame(this.stompSubProtocolHandler, handler); + } + + assertTrue(TestUtils.getPropertyValue(this.customOutboundAdapter, "mergeWithDefaultConverters", Boolean.class)); + CompositeMessageConverter compositeMessageConverter = TestUtils.getPropertyValue(this.customOutboundAdapter, + "messageConverter", CompositeMessageConverter.class); + List converters = compositeMessageConverter.getConverters(); + assertEquals(5, converters.size()); + assertSame(this.simpleMessageConverter, converters.get(0)); + assertSame(this.mapMessageConverter, converters.get(1)); + assertThat(converters.get(2), instanceOf(StringMessageConverter.class)); + assertTrue(TestUtils.getPropertyValue(this.customOutboundAdapter, "client", Boolean.class)); + } + +} diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/server/WebSocketServerTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/server/WebSocketServerTests.java new file mode 100644 index 0000000000..ad74812d59 --- /dev/null +++ b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/server/WebSocketServerTests.java @@ -0,0 +1,249 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.websocket.server; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; + +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.List; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.annotation.Transformer; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; +import org.springframework.integration.websocket.ClientWebSocketContainer; +import org.springframework.integration.websocket.IntegrationWebSocketContainer; +import org.springframework.integration.websocket.JettyWebSocketTestServer; +import org.springframework.integration.websocket.ServerWebSocketContainer; +import org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter; +import org.springframework.integration.websocket.outbound.WebSocketOutboundMessageHandler; +import org.springframework.integration.websocket.support.SubProtocolHandlerRegistry; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler; +import org.springframework.messaging.simp.broker.SubscriptionRegistry; +import org.springframework.messaging.simp.config.MessageBrokerRegistry; +import org.springframework.messaging.simp.stomp.StompCommand; +import org.springframework.messaging.simp.stomp.StompHeaderAccessor; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.MultiValueMap; +import org.springframework.web.socket.client.WebSocketClient; +import org.springframework.web.socket.client.jetty.JettyWebSocketClient; +import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer; +import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; +import org.springframework.web.socket.config.annotation.StompEndpointRegistry; +import org.springframework.web.socket.messaging.StompSubProtocolHandler; +import org.springframework.web.socket.messaging.SubProtocolHandler; +import org.springframework.web.socket.sockjs.client.SockJsClient; +import org.springframework.web.socket.sockjs.client.Transport; +import org.springframework.web.socket.sockjs.client.WebSocketTransport; + +/** + * @author Artem Bilan + * @since 4.1 + */ +@ContextConfiguration(classes = WebSocketServerTests.ContextConfiguration.class) +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class WebSocketServerTests { + + private final static SpelExpressionParser PARSER = new SpelExpressionParser(); + + @Autowired + @Qualifier("webSocketOutputChannel") + private MessageChannel webSocketOutputChannel; + + @Autowired + @Qualifier("webSocketInputChannel") + private PollableChannel webSocketInputChannel; + + @Value("#{server.serverContext.getBean('simpleBrokerMessageHandler')}") + private SimpleBrokerMessageHandler brokerHandler; + + @Test + public void testWebSocketOutboundMessageHandler() throws Exception { + StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SUBSCRIBE); + headers.setSubscriptionId("subs1"); + headers.setDestination("/queue/foo"); + Message message = MessageBuilder.withPayload(ByteBuffer.allocate(0).array()).setHeaders(headers).build(); + + headers = StompHeaderAccessor.create(StompCommand.SEND); + headers.setSubscriptionId("subs1"); + Message message2 = MessageBuilder.withPayload("Spring").setHeaders(headers).build(); + + this.webSocketOutputChannel.send(message); + this.webSocketOutputChannel.send(message2); + + Message received = this.webSocketInputChannel.receive(10000); + assertNotNull(received); + StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(received); + assertEquals(StompCommand.MESSAGE.getMessageType(), stompHeaderAccessor.getMessageType()); + + Object receivedPayload = received.getPayload(); + assertThat(receivedPayload, instanceOf(String.class)); + assertEquals("Hello Spring", receivedPayload); + + SubscriptionRegistry subscriptionRegistry = this.brokerHandler.getSubscriptionRegistry(); + headers = StompHeaderAccessor.create(StompCommand.MESSAGE); + headers.setDestination("/queue/foo"); + message = MessageBuilder.withPayload(ByteBuffer.allocate(0).array()).setHeaders(headers).build(); + MultiValueMap subscriptions = subscriptionRegistry.findSubscriptions(message); + assertFalse(subscriptions.isEmpty()); + List subscription = subscriptions.values().iterator().next(); + assertEquals(1, subscription.size()); + assertEquals("subs1", subscription.get(0)); + } + + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + public JettyWebSocketTestServer server() { + return new JettyWebSocketTestServer(ServerConfig.class); + } + + @Bean + public WebSocketClient webSocketClient() { + return new SockJsClient(Collections.singletonList(new WebSocketTransport(new JettyWebSocketClient()))); + } + + @Bean + public IntegrationWebSocketContainer clientWebSocketContainer() { + return new ClientWebSocketContainer(webSocketClient(), server().getWsBaseUrl() + "/ws"); + } + + @Bean + public SubProtocolHandler stompSubProtocolHandler() { + return new StompSubProtocolHandler(); + } + + @Bean + public PollableChannel webSocketInputChannel() { + return new QueueChannel(); + } + + @Bean + public MessageChannel webSocketOutputChannel() { + return new DirectChannel(); + } + + @Bean + public MessageProducer webSocketInboundChannelAdapter() { + WebSocketInboundChannelAdapter webSocketInboundChannelAdapter = + new WebSocketInboundChannelAdapter(clientWebSocketContainer(), + new SubProtocolHandlerRegistry(stompSubProtocolHandler())); + webSocketInboundChannelAdapter.setOutputChannel(webSocketInputChannel()); + return webSocketInboundChannelAdapter; + } + + @Bean + @ServiceActivator(inputChannel = "webSocketOutputChannel") + public MessageHandler webSocketOutboundMessageHandler() { + return new WebSocketOutboundMessageHandler(clientWebSocketContainer(), + new SubProtocolHandlerRegistry(stompSubProtocolHandler())); + } + + } + + // WebSocket Server part + + @Configuration + @EnableIntegration + @EnableWebSocketMessageBroker + static class ServerConfig extends AbstractWebSocketMessageBrokerConfigurer { + + @Override + public void registerStompEndpoints(StompEndpointRegistry registry) { + registry.addEndpoint("/foo"); + } + + @Override + public void configureMessageBroker(MessageBrokerRegistry registry) { + registry.setApplicationDestinationPrefixes("/app/") + .enableSimpleBroker("/queue/", "/topic/"); + } + + @Bean + public ServerWebSocketContainer serverWebSocketContainer() { + return new ServerWebSocketContainer("/ws").withSockJs(); + } + + @Bean + public SubProtocolHandler stompSubProtocolHandler() { + return new StompSubProtocolHandler(); + } + + @Bean + public MessageChannel webSocketInputChannel() { + return new DirectChannel(); + } + + @Bean + public MessageChannel webSocketOutputChannel() { + return new DirectChannel(); + } + + @Bean + public MessageProducer webSocketInboundChannelAdapter() { + WebSocketInboundChannelAdapter webSocketInboundChannelAdapter = + new WebSocketInboundChannelAdapter(serverWebSocketContainer(), + new SubProtocolHandlerRegistry(stompSubProtocolHandler())); + webSocketInboundChannelAdapter.setOutputChannel(webSocketInputChannel()); + webSocketInboundChannelAdapter.setUseBroker(true); + return webSocketInboundChannelAdapter; + } + + @Bean + @Transformer(inputChannel = "webSocketInputChannel", outputChannel = "webSocketOutputChannel") + public ExpressionEvaluatingTransformer transformer() { + return new ExpressionEvaluatingTransformer(PARSER.parseExpression("'Hello ' + payload")); + } + + @Bean + @ServiceActivator(inputChannel = "webSocketOutputChannel") + public MessageHandler webSocketOutboundMessageHandler() { + return new WebSocketOutboundMessageHandler(serverWebSocketContainer(), + new SubProtocolHandlerRegistry(stompSubProtocolHandler())); + } + + } + +} diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 414569e320..bc4ecc57a1 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -18,6 +18,15 @@ See . +

+ WebSocket support + + The WebSocket module is now available. It is fully based on the Spring WebSocket + and Spring Messaging modules and provides an <inbound-channel-adapter> and an + <outbound-channel-adapter>. + More documentation to follow. + +
General Changes