### Question:
VertxWebSocketSession extends AbstractWebSocketSession<WebSocketBase> { @Override public Mono<Void> send(Publisher<WebSocketMessage> messages) { return Mono.create(sink -> { logger.debug("{}Subscribing to messages publisher", getLogPrefix()); Subscriber<WebSocketMessage> subscriber = new WriteStreamSubscriber.Builder<WebSocketBase, WebSocketMessage>() .writeStream(getDelegate()) .nextHandler(this::messageHandler) .endHook(sink) .requestLimit(requestLimit) .build(); messages.subscribe(subscriber); }); } 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);  }### Answer:
@Test public void shouldSendTextMessage() { TestPublisher<WebSocketMessage> source = TestPublisher.create(); Mono<Void> result = session.send(source); StepVerifier.create(result) .expectSubscription() .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.TEXT, "test1"))) .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.TEXT, "test2"))) .then(() -> source.assertMinRequested(1)) .then(source::complete) .verifyComplete(); verify(mockServerWebSocket).writeTextMessage("test1"); verify(mockServerWebSocket).writeTextMessage("test2"); }
@Test public void shouldSendBinaryMessage() { TestPublisher<WebSocketMessage> source = TestPublisher.create(); Mono<Void> result = session.send(source); StepVerifier.create(result) .expectSubscription() .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.BINARY, "test1"))) .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.BINARY, "test2"))) .then(() -> source.assertMinRequested(1)) .then(source::complete) .verifyComplete(); verify(mockServerWebSocket).writeBinaryMessage(Buffer.buffer("test1")); verify(mockServerWebSocket).writeBinaryMessage(Buffer.buffer("test2")); }
@Test public void shouldSendPingMessage() { TestPublisher<WebSocketMessage> source = TestPublisher.create(); Mono<Void> result = session.send(source); StepVerifier.create(result) .expectSubscription() .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.PING, "test1"))) .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.PING, "test2"))) .then(() -> source.assertMinRequested(1)) .then(source::complete) .verifyComplete(); verify(mockServerWebSocket).writePing(Buffer.buffer("test1")); verify(mockServerWebSocket).writePing(Buffer.buffer("test2")); }
@Test public void shouldSendPongMessage() { TestPublisher<WebSocketMessage> source = TestPublisher.create(); Mono<Void> result = session.send(source); StepVerifier.create(result) .expectSubscription() .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.PONG, "test1"))) .then(() -> source.assertMinRequested(1)) .then(() -> source.next(getWebSocketMessage(WebSocketMessage.Type.PONG, "test2"))) .then(() -> source.assertMinRequested(1)) .then(source::complete) .verifyComplete(); verify(mockServerWebSocket).writePong(Buffer.buffer("test1")); verify(mockServerWebSocket).writePong(Buffer.buffer("test2")); }