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

Conflicts:
	spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
Resolved.
This commit is contained in:
Artem Bilan
2016-05-11 13:22:41 -04:00
committed by Gary Russell
parent 2fa55e7acb
commit df6885e6ae
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

@@ -53,7 +53,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;
@@ -63,6 +63,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);
@@ -107,18 +109,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;
}
@@ -130,6 +145,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();
@@ -148,6 +172,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();
}
@@ -177,7 +203,8 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
private final boolean syncClientLifecycle;
public 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());
@@ -188,6 +215,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
if (this.syncClientLifecycle) {
((Lifecycle) this.client).start();
}
ClientWebSocketContainer.this.connecting = true;
super.startInternal();
}
@@ -201,14 +229,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());
@@ -228,6 +256,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,12 +25,15 @@ 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;
@@ -38,9 +41,13 @@ import org.junit.BeforeClass;
import org.junit.Ignore;
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;
@@ -66,7 +73,28 @@ public class ClientWebSocketContainerTests {
@Test
@Ignore("Ignored until fix for https://bz.apache.org/bugzilla/show_bug.cgi?id=58624")
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));
@@ -104,8 +132,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

@@ -130,7 +130,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");
}
@Before
@@ -145,7 +145,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));
@@ -161,7 +161,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
@@ -177,7 +177,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));
@@ -195,7 +195,7 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
this.webSocketOutputChannel.send(message2);
receive = webSocketInputChannel.receive(10000);
receive = webSocketInputChannel.receive(20000);
assertNotNull(receive);
assertEquals("6", receive.getPayload());
}
@@ -221,7 +221,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());
}
@@ -240,7 +240,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);
@@ -276,7 +276,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);
@@ -308,7 +308,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());
}