@Test public void shouldReceiveTextMessages() { given(mockServerWebSocket.textMessageHandler(any())).will(invocation -> { Handler<String> handler = invocation.getArgument(0); handler.handle("test1"); handler.handle("test2"); return mockServerWebSocket; }); StepVerifier.create(session.receive()) .expectNext(getWebSocketMessage(WebSocketMessage.Type.TEXT, "test1")) .expectNext(getWebSocketMessage(WebSocketMessage.Type.TEXT, "test2")) .verifyComplete(); }
@Override public Flux<WebSocketMessage> receive() { return Flux.create(sink -> { logger.debug("{}Connecting to a web socket read stream", getLogPrefix()); WebSocketBase socket = getDelegate(); socket.pause() .textMessageHandler(payload -> { logger.debug("{}Received text '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(textMessage(payload)); }) .binaryMessageHandler(payload -> { logger.debug("{}Received binary '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(binaryMessage(payload)); }) .pongHandler(payload -> { logger.debug("{}Received pong '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(pongMessage(payload)); }) .exceptionHandler(throwable -> { logger.debug("{}Received exception '{}' from a web socket read stream", getLogPrefix(), throwable); sink.error(throwable); }) .endHandler(e -> { logger.debug("{}Web socket read stream ended", getLogPrefix()); sink.complete(); }); sink.onRequest(i -> { logger.debug("{}Fetching '{}' entries from a web socket read stream", getLogPrefix(), i); socket.fetch(i); }); } ); }
VertxWebSocketSession extends AbstractWebSocketSession<WebSocketBase> { @Override public Flux<WebSocketMessage> receive() { return Flux.create(sink -> { logger.debug("{}Connecting to a web socket read stream", getLogPrefix()); WebSocketBase socket = getDelegate(); socket.pause() .textMessageHandler(payload -> { logger.debug("{}Received text '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(textMessage(payload)); }) .binaryMessageHandler(payload -> { logger.debug("{}Received binary '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(binaryMessage(payload)); }) .pongHandler(payload -> { logger.debug("{}Received pong '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(pongMessage(payload)); }) .exceptionHandler(throwable -> { logger.debug("{}Received exception '{}' from a web socket read stream", getLogPrefix(), throwable); sink.error(throwable); }) .endHandler(e -> { logger.debug("{}Web socket read stream ended", getLogPrefix()); sink.complete(); }); sink.onRequest(i -> { logger.debug("{}Fetching '{}' entries from a web socket read stream", getLogPrefix(), i); socket.fetch(i); }); } ); } }
VertxWebSocketSession extends AbstractWebSocketSession<WebSocketBase> { @Override public Flux<WebSocketMessage> receive() { return Flux.create(sink -> { logger.debug("{}Connecting to a web socket read stream", getLogPrefix()); WebSocketBase socket = getDelegate(); socket.pause() .textMessageHandler(payload -> { logger.debug("{}Received text '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(textMessage(payload)); }) .binaryMessageHandler(payload -> { logger.debug("{}Received binary '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(binaryMessage(payload)); }) .pongHandler(payload -> { logger.debug("{}Received pong '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(pongMessage(payload)); }) .exceptionHandler(throwable -> { logger.debug("{}Received exception '{}' from a web socket read stream", getLogPrefix(), throwable); sink.error(throwable); }) .endHandler(e -> { logger.debug("{}Web socket read stream ended", getLogPrefix()); sink.complete(); }); sink.onRequest(i -> { logger.debug("{}Fetching '{}' entries from a web socket read stream", getLogPrefix(), i); socket.fetch(i); }); } ); } VertxWebSocketSession(WebSocketBase delegate, HandshakeInfo handshakeInfo, BufferConverter bufferConverter,
        int maxWebSocketFrameSize, int maxWebSocketMessageSize); }
VertxWebSocketSession extends AbstractWebSocketSession<WebSocketBase> { @Override public Flux<WebSocketMessage> receive() { return Flux.create(sink -> { logger.debug("{}Connecting to a web socket read stream", getLogPrefix()); WebSocketBase socket = getDelegate(); socket.pause() .textMessageHandler(payload -> { logger.debug("{}Received text '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(textMessage(payload)); }) .binaryMessageHandler(payload -> { logger.debug("{}Received binary '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(binaryMessage(payload)); }) .pongHandler(payload -> { logger.debug("{}Received pong '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(pongMessage(payload)); }) .exceptionHandler(throwable -> { logger.debug("{}Received exception '{}' from a web socket read stream", getLogPrefix(), throwable); sink.error(throwable); }) .endHandler(e -> { logger.debug("{}Web socket read stream ended", getLogPrefix()); sink.complete(); }); sink.onRequest(i -> { logger.debug("{}Fetching '{}' entries from a web socket read stream", getLogPrefix(), i); socket.fetch(i); }); } ); } VertxWebSocketSession(WebSocketBase delegate, HandshakeInfo handshakeInfo, BufferConverter bufferConverter,
        int maxWebSocketFrameSize, int maxWebSocketMessageSize); @Override Flux<WebSocketMessage> receive(); @Override Mono<Void> send(Publisher<WebSocketMessage> messages); @Override Mono<Void> close(CloseStatus status); }
VertxWebSocketSession extends AbstractWebSocketSession<WebSocketBase> { @Override public Flux<WebSocketMessage> receive() { return Flux.create(sink -> { logger.debug("{}Connecting to a web socket read stream", getLogPrefix()); WebSocketBase socket = getDelegate(); socket.pause() .textMessageHandler(payload -> { logger.debug("{}Received text '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(textMessage(payload)); }) .binaryMessageHandler(payload -> { logger.debug("{}Received binary '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(binaryMessage(payload)); }) .pongHandler(payload -> { logger.debug("{}Received pong '{}' from a web socket read stream", getLogPrefix(), payload); sink.next(pongMessage(payload)); }) .exceptionHandler(throwable -> { logger.debug("{}Received exception '{}' from a web socket read stream", getLogPrefix(), throwable); sink.error(throwable); }) .endHandler(e -> { logger.debug("{}Web socket read stream ended", getLogPrefix()); sink.complete(); }); sink.onRequest(i -> { logger.debug("{}Fetching '{}' entries from a web socket read stream", getLogPrefix(), i); socket.fetch(i); }); } ); } VertxWebSocketSession(WebSocketBase delegate, HandshakeInfo handshakeInfo, BufferConverter bufferConverter,
        int maxWebSocketFrameSize, int maxWebSocketMessageSize); @Override Flux<WebSocketMessage> receive(); @Override Mono<Void> send(Publisher<WebSocketMessage> messages); @Override Mono<Void> close(CloseStatus status);  }