GH-8600: Fix WebSocket removeRegistration (#8601)

* GH-8600: Fix WebSocket `removeRegistration`

Fixes https://github.com/spring-projects/spring-integration/issues/8600

When we register a dynamic WebSocket endpoint and use a `WebSocketHandlerDecoratorFactory`
such an endpoint is not removed on an `IntegrationFlow` destruction.
The actual `WebSocketHandler` is decorated, however we still use an initial one
for condition.

* Refactor `IntegrationWebSocketContainer` to expose a `protected` setter for the
`WebSocketHandler` which is called from the `ServerWebSocketContainer` after decoration

**Cherry-pick to `6.0.x` & `5.5.x`**

* Fix language in Javadocs

Co-authored-by: Gary Russell <grussell@vmware.com>

---------

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2023-04-19 12:54:00 -04:00
committed by Gary Russell
parent c44c2e6545
commit d89badf236
4 changed files with 31 additions and 8 deletions

View File

@@ -197,7 +197,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
* <p>
* Opened {@link WebSocketSession} is populated to the wrapping {@link ClientWebSocketContainer}.
* <p>
* The {@link #webSocketHandler} is used to handle {@link WebSocketSession} events.
* The {@link #getWebSocketHandler()} is used to handle {@link WebSocketSession} events.
*/
private final class IntegrationWebSocketConnectionManager extends ConnectionManagerSupport {
@@ -242,8 +242,7 @@ public final class ClientWebSocketContainer extends IntegrationWebSocketContaine
}
ClientWebSocketContainer.this.headers.setSecWebSocketProtocol(getSubProtocols());
ListenableFuture<WebSocketSession> future =
this.client.doHandshake(ClientWebSocketContainer.this.webSocketHandler,
ClientWebSocketContainer.this.headers, getUri());
this.client.doHandshake(getWebSocketHandler(), ClientWebSocketContainer.this.headers, getUri());
future.addCallback(new ListenableFutureCallback<WebSocketSession>() {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2021 the original author or authors.
* Copyright 2014-2023 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.
@@ -67,7 +67,7 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR
protected final WebSocketHandler webSocketHandler = new IntegrationWebSocketHandler(); // NOSONAR
private WebSocketHandler webSocketHandler = new IntegrationWebSocketHandler();
protected final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>(); // NOSONAR
@@ -104,6 +104,15 @@ public abstract class IntegrationWebSocketContainer implements DisposableBean {
}
}
/**
* Replace the default {@link WebSocketHandler} with the one provided here, e.g. via decoration factories.
* @param handler the actual {@link WebSocketHandler} to replace.
* @since 5.5.18
*/
protected void setWebSocketHandler(WebSocketHandler handler) {
this.webSocketHandler = handler;
}
public WebSocketHandler getWebSocketHandler() {
return this.webSocketHandler;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-2023 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.
@@ -150,11 +150,12 @@ public class ServerWebSocketContainer extends IntegrationWebSocketContainer
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
WebSocketHandler webSocketHandler = this.webSocketHandler;
WebSocketHandler webSocketHandler = getWebSocketHandler();
if (this.decoratorFactories != null) {
for (WebSocketHandlerDecoratorFactory factory : this.decoratorFactories) {
webSocketHandler = factory.decorate(webSocketHandler);
setWebSocketHandler(webSocketHandler);
}
}

View File

@@ -16,6 +16,8 @@
package org.springframework.integration.websocket.dsl;
import java.util.concurrent.atomic.AtomicReference;
import javax.websocket.DeploymentException;
import org.junit.jupiter.api.Test;
@@ -39,6 +41,7 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.server.HandshakeHandler;
@@ -48,6 +51,9 @@ import org.springframework.web.socket.server.support.DefaultHandshakeHandler;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
@SpringJUnitConfig(classes = WebSocketDslTests.ClientConfig.class)
@DirtiesContext
@@ -63,13 +69,19 @@ public class WebSocketDslTests {
IntegrationFlowContext integrationFlowContext;
@Test
void testDynamicServerEndpointRegistration() {
void testDynamicServerEndpointRegistration() throws Exception {
// Dynamic server flow
AnnotationConfigWebApplicationContext serverContext = this.server.getServerContext();
IntegrationFlowContext serverIntegrationFlowContext = serverContext.getBean(IntegrationFlowContext.class);
AtomicReference<WebSocketHandler> decoratedHandler = new AtomicReference<>();
ServerWebSocketContainer serverWebSocketContainer =
new ServerWebSocketContainer("/dynamic")
.setHandshakeHandler(serverContext.getBean(HandshakeHandler.class))
.setDecoratorFactories(handler -> {
WebSocketHandler spy = spy(handler);
decoratedHandler.set(spy);
return spy;
})
.withSockJs();
WebSocketInboundChannelAdapter webSocketInboundChannelAdapter =
@@ -108,6 +120,8 @@ public class WebSocketDslTests {
.extracting(Message::getPayload)
.isEqualTo("dynamic test");
verify(decoratedHandler.get()).handleMessage(any(), any());
dynamicServerFlow.destroy();
await() // Looks like endpoint is removed on the server side somewhat async