SpringCloud/spring-cloud-gateway
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch '3.0.x'

Browse Source
This commit is contained in:
spencergibb
2021-11-01 15:40:25 -04:00
parent cee9dee75f 0be612a572
commit 72d4ae09b2
2 changed files with 37 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-cloud-gateway-server/src/main/java/org/springframework/cloud/gateway/filter/WebsocketRoutingFilter.java
View File

@@ -203,6 +203,10 @@ public class WebsocketRoutingFilter implements GlobalFilter, Ordered {
return client.execute(url, this.headers, new WebSocketHandler() {
@Override
public Mono<Void> handle(WebSocketSession proxySession) {
Mono<Void> serverClose = proxySession.closeStatus().filter(__ -> session.isOpen())
.flatMap(session::close);
Mono<Void> proxyClose = session.closeStatus().filter(__ -> proxySession.isOpen())
.flatMap(proxySession::close);
// Use retain() for Reactor Netty
Mono<Void> proxySessionSend = proxySession
.send(session.receive().doOnNext(WebSocketMessage::retain));
@@ -210,6 +214,9 @@ public class WebsocketRoutingFilter implements GlobalFilter, Ordered {
Mono<Void> serverSessionSend = session
.send(proxySession.receive().doOnNext(WebSocketMessage::retain));
// .log("sessionSend", Level.FINE);
// Ensure closeStatus from one propagates to the other
Mono.when(serverClose, proxyClose).subscribe();
// Complete when both sessions are done
return Mono.zip(proxySessionSend, serverSessionSend).then();
}

35
spring-cloud-gateway-server/src/test/java/org/springframework/cloud/gateway/test/websocket/WebSocketIntegrationTests.java
View File

@@ -36,6 +36,7 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.ReplayProcessor;
import reactor.core.publisher.Sinks;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -101,6 +102,8 @@ public class WebSocketIntegrationTests {
private int gatewayPort;
private static final Sinks.One<CloseStatus> serverCloseStatusSink = Sinks.one();
private static Mono<Void> doSend(WebSocketSession session, Publisher<WebSocketMessage> output) {
return session.send(output);
// workaround for suspected RxNetty WebSocket client issue
@@ -241,13 +244,25 @@ public class WebSocketIntegrationTests {
}
@Test
public void sessionClosing() throws Exception {
this.client.execute(getUrl("/close"), session -> {
public void serverClosing() throws Exception {
AtomicReference<Mono<CloseStatus>> closeStatus = new AtomicReference<>();
this.client.execute(getUrl("/server-close"), session -> {
logger.debug("Starting..");
closeStatus.set(session.closeStatus());
return session.receive().doOnNext(s -> logger.debug("inbound " + s)).then().doFinally(signalType -> {
logger.debug("Completed with: " + signalType);
});
}).block(Duration.ofMillis(5000));
assertThat(closeStatus.get().block(Duration.ofMillis(5000)))
.isEqualTo(CloseStatus.create(4999, "server-close"));
}
@Test
public void clientClosing() throws Exception {
this.client.execute(getUrl("/client-close"), session -> session.close(CloseStatus.create(4999, "client-close")))
.block(Duration.ofMillis(5000));
assertThat(serverCloseStatusSink.asMono().block(Duration.ofMillis(5000)))
.isEqualTo(CloseStatus.create(4999, "client-close"));
}
@Configuration(proxyBeanMethods = false)
@@ -279,7 +294,8 @@ public class WebSocketIntegrationTests {
map.put("/echoForHttp", new EchoWebSocketHandler());
map.put("/sub-protocol", new SubProtocolWebSocketHandler());
map.put("/custom-header", new CustomHeaderHandler());
map.put("/close", new SessionClosingHandler());
map.put("/server-close", new ServerClosingHandler());
map.put("/client-close", new ClientClosingHandler());
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setUrlMap(map);
@@ -334,11 +350,20 @@ public class WebSocketIntegrationTests {
}
private static class SessionClosingHandler implements WebSocketHandler {
private static class ServerClosingHandler implements WebSocketHandler {
@Override
public Mono<Void> handle(WebSocketSession session) {
return Flux.never().mergeWith(session.close(CloseStatus.GOING_AWAY)).then();
return Flux.never().mergeWith(session.close(CloseStatus.create(4999, "server-close"))).then();
}
}
private static class ClientClosingHandler implements WebSocketHandler {
@Override
public Mono<Void> handle(WebSocketSession session) {
return session.closeStatus().doOnNext(serverCloseStatusSink::tryEmitValue).then();
}
}