INT-1198: WebSockets: Server Side
JIRA: https://jira.spring.io/browse/INT-1198 INT-1198: WebSocket: Add namespace support Polishing components according to the namespace support experience Parser for `<server-container>` and tests INT-1198: Add parser tests for `<int-websocket:outbound-channel-adapter>` Introduce `use-broker` on server side Polishing according PR comments `What's New` note
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
3d32acf78c
commit
b413239a83
@@ -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<String> getSubProtocols() {
|
||||
return Collections.singletonList("v10.stomp");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -52,7 +52,7 @@ public class JettyWebSocketTestServer implements InitializingBean, DisposableBea
|
||||
}
|
||||
|
||||
public AnnotationConfigWebApplicationContext getServerContext() {
|
||||
return serverContext;
|
||||
return this.serverContext;
|
||||
}
|
||||
|
||||
public String getWsBaseUrl() {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns:int="http://www.springframework.org/schema/integration"
|
||||
xmlns:int-websocket="http://www.springframework.org/schema/integration/websocket"
|
||||
xmlns:websocket="http://www.springframework.org/schema/websocket"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns="http://www.springframework.org/schema/beans"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/beans
|
||||
http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/websocket
|
||||
http://www.springframework.org/schema/websocket/spring-websocket.xsd
|
||||
http://www.springframework.org/schema/integration
|
||||
http://www.springframework.org/schema/integration/spring-integration.xsd
|
||||
http://www.springframework.org/schema/integration/websocket
|
||||
http://www.springframework.org/schema/integration/websocket/spring-integration-websocket.xsd">
|
||||
|
||||
<int-websocket:server-container id="serverWebSocketContainer"
|
||||
path="/ws"
|
||||
send-buffer-size-limit="100000"
|
||||
send-time-limit="100"
|
||||
handshake-handler="handshakeHandler"
|
||||
handshake-interceptors="handshakeInterceptor">
|
||||
<int-websocket:sockjs client-library-url="https://foo.sock.js"
|
||||
disconnect-delay="4000"
|
||||
heartbeat-time="30000"
|
||||
message-cache-size="10000"
|
||||
session-cookie-needed="false"
|
||||
stream-bytes-limit="2000"
|
||||
websocket-enabled="false"
|
||||
scheduler="taskScheduler"
|
||||
message-codec="sockJsMessageCodec"/>
|
||||
</int-websocket:server-container>
|
||||
|
||||
<bean id="handshakeHandler" class="org.mockito.Mockito" factory-method="mock">
|
||||
<constructor-arg value="org.springframework.web.socket.server.HandshakeHandler"/>
|
||||
</bean>
|
||||
|
||||
<bean id="handshakeInterceptor" class="org.mockito.Mockito" factory-method="mock">
|
||||
<constructor-arg value="org.springframework.web.socket.server.HandshakeInterceptor"/>
|
||||
</bean>
|
||||
|
||||
<bean id="sockJsMessageCodec" class="org.mockito.Mockito" factory-method="mock">
|
||||
<constructor-arg value="org.springframework.web.socket.sockjs.frame.SockJsMessageCodec"/>
|
||||
</bean>
|
||||
|
||||
<websocket:message-broker application-destination-prefix="/app">
|
||||
<websocket:stomp-endpoint path="/foo"/>
|
||||
<websocket:simple-broker prefix="/topic"/>
|
||||
</websocket:message-broker>
|
||||
|
||||
<int-websocket:inbound-channel-adapter id="defaultInboundAdapter" container="serverWebSocketContainer"
|
||||
use-broker="true"/>
|
||||
|
||||
<bean id="webSocketClient" class="org.mockito.Mockito" factory-method="mock">
|
||||
<constructor-arg value="org.springframework.web.socket.client.WebSocketClient"/>
|
||||
</bean>
|
||||
|
||||
<int-websocket:client-container id="clientWebSocketContainer"
|
||||
client="webSocketClient"
|
||||
uri="ws://foo.bar/{var}?service={service}"
|
||||
uri-variables="ws,user"
|
||||
send-buffer-size-limit="1000"
|
||||
send-time-limit="100"
|
||||
origin="FOO"
|
||||
phase="100">
|
||||
<int-websocket:http-headers>
|
||||
<entry key="FOO" value="BAR,baz"/>
|
||||
</int-websocket:http-headers>
|
||||
</int-websocket:client-container>
|
||||
|
||||
<bean id="stompSubProtocolHandler" class="org.springframework.web.socket.messaging.StompSubProtocolHandler"/>
|
||||
|
||||
<bean id="passThruSubProtocolHandler"
|
||||
class="org.springframework.integration.websocket.support.PassThruSubProtocolHandler"/>
|
||||
|
||||
<bean id="simpleMessageConverter" class="org.springframework.integration.support.converter.SimpleMessageConverter"/>
|
||||
|
||||
<bean id="mapMessageConverter" class="org.springframework.integration.support.converter.MapMessageConverter"/>
|
||||
|
||||
<int-websocket:inbound-channel-adapter id="customInboundAdapter" container="clientWebSocketContainer"
|
||||
auto-startup="false"
|
||||
payload-type="java.lang.Integer"
|
||||
default-protocol-handler="stompSubProtocolHandler"
|
||||
protocol-handlers="passThruSubProtocolHandler"
|
||||
message-converters="simpleMessageConverter,mapMessageConverter"
|
||||
merge-with-default-converters="true"
|
||||
channel="clientInboundChannel"
|
||||
error-channel="errorChannel"
|
||||
send-timeout="2000"
|
||||
phase="200"/>
|
||||
|
||||
<int:channel id="clientInboundChannel"/>
|
||||
|
||||
<int-websocket:outbound-channel-adapter id="defaultOutboundAdapter" container="serverWebSocketContainer"/>
|
||||
|
||||
<int-websocket:outbound-channel-adapter id="customOutboundAdapter" container="clientWebSocketContainer"
|
||||
default-protocol-handler="stompSubProtocolHandler"
|
||||
protocol-handlers="passThruSubProtocolHandler"
|
||||
message-converters="simpleMessageConverter,mapMessageConverter"
|
||||
merge-with-default-converters="true"
|
||||
channel="clientOutboundChannel"/>
|
||||
|
||||
<int:channel id="clientOutboundChannel"/>
|
||||
|
||||
</beans>
|
||||
@@ -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<TransportType, TransportHandler> 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<MessageConverter> 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<MessageConverter> 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));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<byte[]> message = MessageBuilder.withPayload(ByteBuffer.allocate(0).array()).setHeaders(headers).build();
|
||||
|
||||
headers = StompHeaderAccessor.create(StompCommand.SEND);
|
||||
headers.setSubscriptionId("subs1");
|
||||
Message<String> 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<String, String> subscriptions = subscriptionRegistry.findSubscriptions(message);
|
||||
assertFalse(subscriptions.isEmpty());
|
||||
List<String> 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.<Transport>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()));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user