Finally, one topic that should be covered here is the verification of streaming systems. In this section, we will cover the testing of the WebSocket server and client only. As we will learn in Chapter 6, WebFlux Async Non-Blocking Communication, along with the WebSocket API, there is a Server-Sent Events (SSE) protocol for streaming data, which gives us similar capabilities. Nevertheless, since the implementation of SSE is almost identical to the implementation of the regular controller, all verification techniques from the previous section will be valid for that case as well. Consequently, the only thing that remains unclear now is how to test WebSocket.

Unfortunately, WebFlux does not provide an out-of-the-box solution for WebSocket API testing. Nonetheless, we may use a standard tool set for building the verification classes. Namely, we may use WebSocketClient to connect to the target server and verify the correctness of the received data. The following code depicts such an approach:

new ReactorNettyWebSocketClient()
      .execute(uri, new WebSocketHandler() {...})

Despite the fact that we may connect to the server, it is difficult to verify incoming data using StepVerifier. First of all, .execute() returns Mono<Void> instead of incoming data from the WebSocket connection. In turn, we need to check the two side interactions, which means that there are cases in which it is important to check that incoming data is the result of outgoing messages. An example of such a system might be a trading platform. Suppose that we have a crypto-trading platform that offers the ability to send trades and receive the results of the deals. One of the business requirements is the ability to make Bitcoin trades. This means that the user may sell or buy bitcoins using the platform and observe the result of the deal. To verify the functionality, we need to check that incoming trade is the result of the outgoing request. From a testing perspective, it is hard to deal with WebSocketHandler to check all corner cases. Consequently, from a testing perspective, the WebSocket client's interface would ideally look as follows:

interface TestWebSocketClient {
   Flux<WebSocketMessage> sendAndReceive(Publisher<?> outgoingSource);
}

To adapt the standard WebSocketClient to the proposed TestWebSocketClient, we need to go through the following steps.

First, we need to handle WebSocketSession, given in WebSocketHandler via Mono<WebSocketSession>, as shown in the following code:

Mono.create(sink ->
   sink.onCancel(
         client.execute(uri, session -> {
                    sink.success(session);
                    return Mono.never();
               })
               .doOnError(sink::error)
               .subscribe()
    )
);

Using Mono.create() and MonoSink, we may adopt the old school way of handling asynchronous callbacks with a session and redirect it to another stream. In turn, we need to care about the correct return type for the WebSocketHandler#handle method. This is because the return type controls the time of the opened connection's life. On the other hand, the connection should be cancelled as soon as MonoSink notifies us about this. Consequently, Mono.never() is the best candidate, which, in combination with the redirection error via .doOnError(sink::error) and the handling cancellation via sink.onCancel() , makes that adoption complete.

The second step that should be performed regarding the proposed API is adapting the WebSocketSession using the following technique:

public Flux<WebSocketMessage> sendAndReceive(
   Publisher<?> outgoingSource
) {
   ...
   .flatMapMany(session ->
       session.receive()
              .mergeWith(
                 Flux.from(outgoingSource)
                     .map(Object::toString)
                     .map(session::textMessage)
                     .as(session::send)
                     .then(Mono.empty())
               )
    );
}

Here, we downstream incoming WebSocketMessages and send outgoing messages to the server. As we might have noticed, in this example, we use plain object conversion, which might be replaced with sophisticated message mapping.

Finally, using that API, we may build the following verification's flow for the mentioned functionality:

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = 
                SpringBootTest.WebEnvironment.DEFINED_PORT)
public class WebSocketAPITests {
   @Test
   @WithMockUser
   public void checkThatUserIsAbleToMakeATrade() {
      URI uri = URI.create("ws://localhost:8080/stream");
      TestWebSocketClient client = TestWebSocketClient.create(uri);
      TestPublisher<String> testPublisher = TestPublisher.create();
      Flux<String> inbound = testPublisher
         .flux()
         .subscribeWith(ReplayProcessor.create(1))
         .transform(client::sendAndReceive)
         .map(WebSocketMessage::getPayloadAsText);

      StepVerifier
         .create(inbound)
         .expectSubscription()
         .then(() -> testPublisher.next("TRADES|BTC"))
         .expectNext("PRICE|AMOUNT|CURRENCY")
         .then(() -> testPublisher.next("TRADE: 10123|1.54|BTC"))
         .expectNext("10123|1.54|BTC")
         .then(() -> testPublisher.next("TRADE: 10090|-0.01|BTC"))
         .expectNext("10090|-0.01|BTC")
         .thenCancel()
         .verify();
    }
}

The first thing we do in this example is configuring the WebEnvironment. By setting the WebEnvironment.DEFINED_PORT, we tell Spring Framework that it should be available on a configured port. This is an essential step because WebSocketClient can connect to the defined handler only over the real HTTP call. We then prepare the inbound stream. In our case, it is important to cache the first message, sent via TestPublisher in the .then() step. This is because .then() may be called before the session can be retrieved, which means that the first message may be ignored, and we may not connect to our bitcoin trades. The next step is to verify that the sent trades have been passed, and that we have received the correct response.

Lastly, it should be mentioned that, along with the WebSocket API verification, there may be cases in which we need to mock interaction over WebSocketClient with external services. Unfortunately, there is no easy way to mock an interaction. First of all, this is because we do not have a common WebSocketClient.Build that might be mocked. In turn, there is no out-of-the-box way to autowire WebSocketClient. Consequently, the only solution that we might have is wiring a mock server.