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

Uses Mono.zip().then() rather than Mono.when().

Browse Source 
fixes gh-260
This commit is contained in:
Spencer Gibb
2018-05-16 17:15:34 -04:00
parent ac3afa3be1
commit 7a1bbf08b3
2 changed files with 26 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/WebsocketRoutingFilter.java
View File

@@ -157,7 +157,7 @@ public class WebsocketRoutingFilter implements GlobalFilter, Ordered {
Mono<Void> serverSessionSend = session
.send(proxySession.receive().doOnNext(WebSocketMessage::retain));
// .log("sessionSend", Level.FINE);
return Mono.when(proxySessionSend, serverSessionSend);
return Mono.zip(proxySessionSend, serverSessionSend).then();
}
/**

25
spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/websocket/WebSocketIntegrationTests.java
View File

@@ -34,6 +34,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import org.springframework.web.reactive.socket.CloseStatus;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
@@ -177,6 +178,7 @@ 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());
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setUrlMap(map);
@@ -287,6 +289,21 @@ public class WebSocketIntegrationTests {
assertEquals("my-header:my-value", output.block(Duration.ofMillis(5000)));
}
@Test
public void sessionClosing() throws Exception {
this.client.execute(getUrl("/close"),
session -> {
logger.debug("Starting..");
return session.receive()
.doOnNext(s -> logger.debug("inbound " + s))
.then()
.doFinally(signalType -> {
logger.debug("Completed with: " + signalType);
});
})
.block(Duration.ofMillis(5000));
}
private static class EchoWebSocketHandler implements WebSocketHandler {
@Override
@@ -330,6 +347,14 @@ public class WebSocketIntegrationTests {
}
}
private static class SessionClosingHandler implements WebSocketHandler {
@Override
public Mono<Void> handle(WebSocketSession session) {
return Flux.never().mergeWith(session.close(CloseStatus.GOING_AWAY)).then();
}
}
private static Mono<Void> doSend(WebSocketSession session, Publisher<WebSocketMessage> output) {
return session.send(output);
// workaround for suspected RxNetty WebSocket client issue