@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")); }
@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 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 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); }
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); }
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);  }