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

Polish WebSocketIntegrationTests

Browse Source
This commit is contained in:
Rossen Stoyanchev
2018-05-01 14:26:21 -04:00
parent 417bb302c3
commit de1eb343e7
2 changed files with 13 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/ReactorNettyWebSocketSession.java
View File

@@ -20,7 +20,6 @@ import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.ipc.netty.NettyInbound; import reactor.ipc.netty.NettyInbound;
import reactor.ipc.netty.NettyOutbound; import reactor.ipc.netty.NettyOutbound;
import reactor.ipc.netty.NettyPipeline; import reactor.ipc.netty.NettyPipeline;
@@ -35,8 +34,8 @@ import org.springframework.web.reactive.socket.WebSocketSession;
/** /**
* Spring {@link WebSocketSession} implementation that adapts to Reactor Netty's * {@link WebSocketSession} implementation for use with the Reactor Netty's
* WebSocket {@link NettyInbound} and {@link NettyOutbound}. * {@link NettyInbound} and {@link NettyOutbound}.
* *
* @author Rossen Stoyanchev * @author Rossen Stoyanchev
* @since 5.0 * @since 5.0
@@ -44,8 +43,6 @@ import org.springframework.web.reactive.socket.WebSocketSession;
public class ReactorNettyWebSocketSession public class ReactorNettyWebSocketSession
extends NettyWebSocketSessionSupport<ReactorNettyWebSocketSession.WebSocketConnection> { extends NettyWebSocketSessionSupport<ReactorNettyWebSocketSession.WebSocketConnection> {
private final MonoProcessor<WebSocketFrame> closeMono = MonoProcessor.create();
public ReactorNettyWebSocketSession(WebsocketInbound inbound, WebsocketOutbound outbound, public ReactorNettyWebSocketSession(WebsocketInbound inbound, WebsocketOutbound outbound,
HandshakeInfo info, NettyDataBufferFactory bufferFactory) { HandshakeInfo info, NettyDataBufferFactory bufferFactory) {

23
spring-webflux/src/test/java/org/springframework/web/reactive/socket/WebSocketIntegrationTests.java
View File

@@ -43,13 +43,14 @@ import static org.junit.Assert.assertThat;
/** /**
* Integration tests with server-side {@link WebSocketHandler}s. * Integration tests with server-side {@link WebSocketHandler}s.
*
* @author Rossen Stoyanchev * @author Rossen Stoyanchev
*/ */
public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests { public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests {
private static final Log logger = LogFactory.getLog(WebSocketIntegrationTests.class); private static final Log logger = LogFactory.getLog(WebSocketIntegrationTests.class);
private static final Duration TIMEOUT = Duration.ofMillis(5000);
@Override @Override
protected Class<?> getWebConfigClass() { protected Class<?> getWebConfigClass() {
@@ -71,14 +72,12 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests
.thenMany(session.receive().take(count).map(WebSocketMessage::getPayloadAsText)) .thenMany(session.receive().take(count).map(WebSocketMessage::getPayloadAsText))
.subscribeWith(output) .subscribeWith(output)
.doOnNext(s -> logger.debug("inbound " + s)) .doOnNext(s -> logger.debug("inbound " + s))
.then() .then();
.doOnSuccessOrError((aVoid, ex) ->
logger.debug("Done with " + (ex != null ? ex.getMessage() : "success")));
}) })
.block(Duration.ofMillis(5000)); .doOnSuccessOrError((aVoid, ex) -> logger.debug("Done: " + (ex != null ? ex.getMessage() : "success")))
.block(TIMEOUT);
assertEquals(input.collectList().block(Duration.ofMillis(5000)), assertEquals(input.collectList().block(TIMEOUT), output.collectList().block(TIMEOUT));
output.collectList().block(Duration.ofMillis(5000)));
} }
@Test @Test
@@ -102,13 +101,13 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests
.then(); .then();
} }
}) })
.block(Duration.ofMillis(5000)); .block(TIMEOUT);
HandshakeInfo info = infoRef.get(); HandshakeInfo info = infoRef.get();
assertThat(info.getHeaders().getFirst("Upgrade"), Matchers.equalToIgnoringCase("websocket")); assertThat(info.getHeaders().getFirst("Upgrade"), Matchers.equalToIgnoringCase("websocket"));
assertEquals(protocol, info.getHeaders().getFirst("Sec-WebSocket-Protocol")); assertEquals(protocol, info.getHeaders().getFirst("Sec-WebSocket-Protocol"));
assertEquals("Wrong protocol accepted", protocol, info.getSubProtocol()); assertEquals("Wrong protocol accepted", protocol, info.getSubProtocol());
assertEquals("Wrong protocol detected on the server side", protocol, output.block(Duration.ofMillis(5000))); assertEquals("Wrong protocol detected on the server side", protocol, output.block(TIMEOUT));
} }
@Test @Test
@@ -122,9 +121,9 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests
.map(WebSocketMessage::getPayloadAsText) .map(WebSocketMessage::getPayloadAsText)
.subscribeWith(output) .subscribeWith(output)
.then()) .then())
.block(Duration.ofMillis(5000)); .block(TIMEOUT);
assertEquals("my-header:my-value", output.block(Duration.ofMillis(5000))); assertEquals("my-header:my-value", output.block(TIMEOUT));
} }
@Test @Test
@@ -139,7 +138,7 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests
logger.debug("Completed with: " + signalType); logger.debug("Completed with: " + signalType);
}); });
}) })
.block(Duration.ofMillis(5000)); .block(TIMEOUT);
} }