diff --git a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java index a0bc2f9230..bb293c869b 100644 --- a/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java +++ b/spring-integration-stomp/src/test/java/org/springframework/integration/stomp/outbound/StompMessageHandlerWebSocketIntegrationTests.java @@ -91,6 +91,7 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport; /** * @author Artem Bilan * @author Gary Russell + * * @since 4.2 */ @ContextConfiguration(classes = StompMessageHandlerWebSocketIntegrationTests.ContextConfiguration.class) @@ -269,7 +270,7 @@ public class StompMessageHandlerWebSocketIntegrationTests extends LogAdjustingTe @Override public void configureClientInboundChannel(ChannelRegistration registration) { - registration.setInterceptors(new ChannelInterceptorAdapter() { + registration.interceptors(new ChannelInterceptorAdapter() { private final AtomicBoolean invoked = new AtomicBoolean(); diff --git a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java index 25dc055b66..08811bf3ca 100644 --- a/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java +++ b/spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ServerWebSocketContainer.java @@ -18,6 +18,8 @@ package org.springframework.integration.websocket; import java.util.Arrays; +import org.springframework.context.Lifecycle; +import org.springframework.context.SmartLifecycle; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; @@ -46,7 +48,8 @@ import org.springframework.web.socket.sockjs.transport.TransportHandler; * @author Gary Russell * @since 4.1 */ -public class ServerWebSocketContainer extends IntegrationWebSocketContainer implements WebSocketConfigurer { +public class ServerWebSocketContainer extends IntegrationWebSocketContainer + implements WebSocketConfigurer, SmartLifecycle { private final String[] paths; @@ -60,6 +63,10 @@ public class ServerWebSocketContainer extends IntegrationWebSocketContainer impl private String[] origins; + private boolean autoStartup = true; + + private int phase = 0; + public ServerWebSocketContainer(String... paths) { this.paths = paths; } @@ -178,6 +185,51 @@ public class ServerWebSocketContainer extends IntegrationWebSocketContainer impl } + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + @Override + public int getPhase() { + return this.phase; + } + + @Override + public boolean isRunning() { + return this.handshakeHandler instanceof Lifecycle && ((Lifecycle) this.handshakeHandler).isRunning(); + } + + @Override + public synchronized void start() { + if (this.handshakeHandler instanceof Lifecycle && !isRunning()) { + ((Lifecycle) this.handshakeHandler).start(); + } + } + + @Override + public void stop() { + if (isRunning()) { + ((Lifecycle) this.handshakeHandler).start(); + } + } + + @Override + public void stop(Runnable callback) { + if (isRunning()) { + ((Lifecycle) this.handshakeHandler).stop(); + } + callback.run(); + } + /** * @see org.springframework.web.socket.config.annotation.SockJsServiceRegistration */ 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 ada3f6944e..3fc50e9afc 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 @@ -61,6 +61,7 @@ import org.springframework.web.socket.messaging.SessionConnectedEvent; /** * @author Artem Bilan + * * @since 4.1 */ public class WebSocketInboundChannelAdapter extends MessageProducerSupport @@ -255,6 +256,9 @@ public class WebSocketInboundChannelAdapter extends MessageProducerSupport @Override protected void doStop() { this.active = false; + if (this.webSocketContainer instanceof Lifecycle) { + ((Lifecycle) this.webSocketContainer).stop(); + } } private boolean isActive() { 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 index 9414f8d7f2..985510680a 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-2017 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. @@ -23,6 +23,8 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.nio.ByteBuffer; import java.util.Collections; @@ -41,6 +43,7 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.ApplicationListener; +import org.springframework.context.Lifecycle; import org.springframework.context.PayloadApplicationEvent; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -86,12 +89,16 @@ import org.springframework.web.socket.handler.WebSocketHandlerDecorator; import org.springframework.web.socket.handler.WebSocketHandlerDecoratorFactory; import org.springframework.web.socket.messaging.StompSubProtocolHandler; import org.springframework.web.socket.messaging.SubProtocolHandler; +import org.springframework.web.socket.server.RequestUpgradeStrategy; +import org.springframework.web.socket.server.standard.TomcatRequestUpgradeStrategy; +import org.springframework.web.socket.server.support.DefaultHandshakeHandler; 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) @@ -115,6 +122,9 @@ public class WebSocketServerTests { @Value("#{server.serverContext.getBean('webSocketEvents')}") private PollableChannel webSocketEvents; + @Value("#{server.serverContext.getBean('requestUpgradeStrategy')}") + private Lifecycle requestUpgradeStrategy; + @Test public void testWebSocketOutboundMessageHandler() throws Exception { StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.SUBSCRIBE); @@ -151,6 +161,8 @@ public class WebSocketServerTests { Message event = this.webSocketEvents.receive(10000); assertNotNull(event); assertThat(event.getPayload(), instanceOf(WebSocketSession.class)); + + verify(this.requestUpgradeStrategy).start(); } @Test @@ -250,9 +262,20 @@ public class WebSocketServerTests { return new TestWebSocketHandlerDecoratorFactory(); } + @Bean + public RequestUpgradeStrategy requestUpgradeStrategy() { + return spy(new MyTomcatRequestUpgradeStrategy()); + } + + @Bean + public DefaultHandshakeHandler handshakeHandler() { + return new DefaultHandshakeHandler(requestUpgradeStrategy()); + } + @Bean public ServerWebSocketContainer serverWebSocketContainer() { return new ServerWebSocketContainer("/ws") + .setHandshakeHandler(handshakeHandler()) .setDecoratorFactories(testWebSocketHandlerDecoratorFactory()) .setAllowedOrigins("http://foo.com") .withSockJs(); @@ -353,4 +376,23 @@ public class WebSocketServerTests { } + public static class MyTomcatRequestUpgradeStrategy extends TomcatRequestUpgradeStrategy implements Lifecycle { + + @Override + public void start() { + + } + + @Override + public void stop() { + + } + + @Override + public boolean isRunning() { + return true; + } + + } + }