From 39ee4ea3b6b1be8480ed158083655536ec0def0c Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 11 May 2016 13:22:41 -0400 Subject: [PATCH] INT-4028: Fix ClientWebSocketContainer start/stop JIRA: https://jira.spring.io/browse/INT-4028 Previously the `ClientWebSocketContainer` after the connection failure couldn't be restored even after `stop()/start()` because the `openConnectionException` property hasn't been clear on `stop()` * Add `openConnectionException = null` to `stopInternal()` logic * Also clear `openConnectionException` and `clientSession` on `start()` * Plus add recovery (restart) logic into the `getSession()` if the `!clientSession.isOpen()` **Cherry-pick to 4.2.x** Fix race condition around `start/stop` The new `start/stop` logic brakes the in-flight connection. * Introduce one more `connecting` flag to indicate that we are in connecting process. * Absorb the a new test for `ClientWebSocketContainer` with the existing one. Looks like there is some extra session close in between. * Adjust timeouts in the `StompIntegrationTests` to 20 secs. Add `org.apache.catalina` category for tracing logs --- .../websocket/ClientWebSocketContainer.java | 45 +++++++++++++--- .../ClientWebSocketContainerTests.java | 54 +++++++++++++++++-- .../client/StompIntegrationTests.java | 18 +++---- 3 files changed, 97 insertions(+), 20 deletions(-) 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 c568031825..9d576db206 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 @@ -54,7 +54,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine private final WebSocketHttpHeaders headers = new WebSocketHttpHeaders(); - private final ConnectionManagerSupport connectionManager; + private final IntegrationWebSocketConnectionManager connectionManager; private volatile CountDownLatch connectionLatch; @@ -64,6 +64,8 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine private volatile int connectionTimeout = DEFAULT_CONNECTION_TIMEOUT; + private volatile boolean connecting; + public ClientWebSocketContainer(WebSocketClient client, String uriTemplate, Object... uriVariables) { Assert.notNull(client, "'client' must not be null"); this.connectionManager = new IntegrationWebSocketConnectionManager(client, uriTemplate, uriVariables); @@ -108,18 +110,31 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine @Override public WebSocketSession getSession(String sessionId) { if (isRunning()) { + if (!isConnected() && !this.connecting) { + stop(); + start(); + } + try { this.connectionLatch.await(this.connectionTimeout, TimeUnit.SECONDS); } catch (InterruptedException e) { logger.error("'clientSession' has not been established during 'openConnection'"); } + this.connecting = false; } - if (this.openConnectionException != null) { - throw new IllegalStateException(this.openConnectionException); + + try { + if (this.openConnectionException != null) { + throw new IllegalStateException(this.openConnectionException); + } + Assert.state(this.clientSession != null, + "'clientSession' has not been established. Consider to 'start' this container."); + } + catch (IllegalStateException e) { + stop(); + throw e; } - Assert.state(this.clientSession != null, - "'clientSession' has not been established. Consider to 'start' this container."); return this.clientSession; } @@ -131,6 +146,15 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine this.connectionManager.setPhase(phase); } + /** + * Return {@code true} if the {@link #clientSession} is opened. + * @return the {@link WebSocketSession#isOpen()} state. + * @since 4.2.6 + */ + public boolean isConnected() { + return this.connectionManager.isConnected(); + } + @Override public boolean isAutoStartup() { return this.connectionManager.isAutoStartup(); @@ -149,6 +173,8 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine @Override public synchronized void start() { if (!isRunning()) { + this.clientSession = null; + this.openConnectionException = null; this.connectionLatch = new CountDownLatch(1); this.connectionManager.start(); } @@ -178,7 +204,8 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine private final boolean syncClientLifecycle; - private IntegrationWebSocketConnectionManager(WebSocketClient client, String uriTemplate, Object... uriVariables) { + private IntegrationWebSocketConnectionManager(WebSocketClient client, String uriTemplate, + Object... uriVariables) { super(uriTemplate, uriVariables); this.client = client; this.syncClientLifecycle = ((client instanceof Lifecycle) && !((Lifecycle) client).isRunning()); @@ -189,6 +216,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine if (this.syncClientLifecycle) { ((Lifecycle) this.client).start(); } + ClientWebSocketContainer.this.connecting = true; super.startInternal(); } @@ -202,14 +230,14 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine } finally { ClientWebSocketContainer.this.clientSession = null; + ClientWebSocketContainer.this.openConnectionException = null; } } @Override protected void openConnection() { - logger.info("Connecting to WebSocket at " + getUri()); - ClientWebSocketContainer.this.headers.setSecWebSocketProtocol(ClientWebSocketContainer.this.getSubProtocols()); + ClientWebSocketContainer.this.headers.setSecWebSocketProtocol(getSubProtocols()); ListenableFuture future = this.client.doHandshake(ClientWebSocketContainer.this.webSocketHandler, ClientWebSocketContainer.this.headers, getUri()); @@ -229,6 +257,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine ClientWebSocketContainer.this.openConnectionException = t; ClientWebSocketContainer.this.connectionLatch.countDown(); } + }); } 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 449311402e..62c918dc8d 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 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. @@ -25,21 +25,28 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.net.URI; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CancellationException; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.apache.tomcat.websocket.WsWebSocketContainer; import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; +import org.springframework.http.HttpHeaders; +import org.springframework.util.concurrent.ListenableFuture; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.PingMessage; import org.springframework.web.socket.PongMessage; +import org.springframework.web.socket.WebSocketExtension; +import org.springframework.web.socket.WebSocketHandler; import org.springframework.web.socket.WebSocketMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.client.standard.StandardWebSocketClient; @@ -64,7 +71,28 @@ public class ClientWebSocketContainerTests { @Test public void testClientWebSocketContainer() throws Exception { - StandardWebSocketClient webSocketClient = new StandardWebSocketClient(); + + final AtomicBoolean failure = new AtomicBoolean(); + + StandardWebSocketClient webSocketClient = new StandardWebSocketClient() { + + @Override + protected ListenableFuture doHandshakeInternal(WebSocketHandler webSocketHandler, + HttpHeaders headers, URI uri, List protocols, List extensions, + Map attributes) { + + ListenableFuture future = + super.doHandshakeInternal(webSocketHandler, headers, uri, protocols, extensions, + attributes); + if (failure.get()) { + future.cancel(true); + } + + return future; + } + + }; + Map userProperties = new HashMap(); userProperties.put(WsWebSocketContainer.IO_TIMEOUT_MS_PROPERTY, "" + (WsWebSocketContainer.IO_TIMEOUT_MS_DEFAULT * 6)); @@ -102,8 +130,28 @@ public class ClientWebSocketContainerTests { assertFalse(session.isOpen()); assertTrue(messageListener.started); assertThat(messageListener.message, instanceOf(PongMessage.class)); - } + failure.set(true); + + container.start(); + + try { + container.getSession(null); + fail("IllegalStateException is expected"); + } + catch (Exception e) { + assertThat(e, instanceOf(IllegalStateException.class)); + assertThat(e.getCause(), instanceOf(CancellationException.class)); + } + + failure.set(false); + + container.start(); + + session = container.getSession(null); + assertNotNull(session); + assertTrue(session.isOpen()); + } private class TestWebSocketListener implements WebSocketListener { 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 f6bbd7bb3f..9ab29d2dc1 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 @@ -128,7 +128,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { private QueueChannel webSocketEvents; public StompIntegrationTests() { - super("org.springframework", "org.springframework.integration"); + super("org.springframework", "org.springframework.integration", "org.apache.catalina"); } @@ -137,7 +137,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT); this.webSocketOutputChannel.send(MessageBuilder.withPayload(new byte[0]).setHeaders(headers).build()); - Message receive = this.webSocketEvents.receive(10000); + Message receive = this.webSocketEvents.receive(20000); assertNotNull(receive); Object event = receive.getPayload(); assertThat(event, instanceOf(SessionConnectedEvent.class)); @@ -153,7 +153,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { this.webSocketOutputChannel.send(message); SimpleController controller = this.serverContext.getBean(SimpleController.class); - assertTrue(controller.latch.await(10, TimeUnit.SECONDS)); + assertTrue(controller.latch.await(20, TimeUnit.SECONDS)); } @Test @@ -169,7 +169,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { this.webSocketOutputChannel.send(message); - Message receive = this.webSocketEvents.receive(10000); + Message receive = this.webSocketEvents.receive(20000); assertNotNull(receive); Object event = receive.getPayload(); assertThat(event, instanceOf(ReceiptEvent.class)); @@ -187,7 +187,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { this.webSocketOutputChannel.send(message2); - receive = webSocketInputChannel.receive(10000); + receive = webSocketInputChannel.receive(20000); assertNotNull(receive); assertEquals("6", receive.getPayload()); } @@ -213,7 +213,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { this.webSocketOutputChannel.send(message2); - Message receive = webSocketInputChannel.receive(10000); + Message receive = webSocketInputChannel.receive(20000); assertNotNull(receive); assertEquals("10", receive.getPayload()); } @@ -232,7 +232,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { this.webSocketOutputChannel.send(message); - Message receive = webSocketInputChannel.receive(10000); + Message receive = webSocketInputChannel.receive(20000); assertNotNull(receive); StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(receive); @@ -268,7 +268,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { this.webSocketOutputChannel.send(message2); - Message receive = webSocketInputChannel.receive(10000); + Message receive = webSocketInputChannel.receive(20000); assertNotNull(receive); StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.wrap(receive); @@ -300,7 +300,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport { this.webSocketOutputChannel.send(message2); - Message receive = webSocketInputChannel.receive(10000); + Message receive = webSocketInputChannel.receive(20000); assertNotNull(receive); assertEquals("Hello Bob", receive.getPayload()); }