Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4028: Fix ClientWebSocketContainer start/stop

Browse Source 
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
This commit is contained in:
Artem Bilan
2016-05-11 13:22:41 -04:00
committed by Gary Russell
parent 06a1d503be
commit 39ee4ea3b6
3 changed files with 97 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
View File

@@ -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<WebSocketSession> 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();
}
});
}

54
spring-integration-websocket/src/test/java/org/springframework/integration/websocket/ClientWebSocketContainerTests.java
View File

@@ -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<WebSocketSession> doHandshakeInternal(WebSocketHandler webSocketHandler,
HttpHeaders headers, URI uri, List<String> protocols, List<WebSocketExtension> extensions,
Map<String, Object> attributes) {
ListenableFuture<WebSocketSession> future =
super.doHandshakeInternal(webSocketHandler, headers, uri, protocols, extensions,
attributes);
if (failure.get()) {
future.cancel(true);
}
return future;
}
};
Map<String, Object> userProperties = new HashMap<String, Object>();
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 {

18
spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java
View File

@@ -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());
}