The first step of the publisher's testing is the verification of the infinite Publisher. According to the Reactive Streams specification, an infinite stream means that the stream will never call the Subscriber#onComplete() method. In turn, this means that the testing techniques that we may have learned before will not work anymore. The problem here is that StepVerifier is going to wait for the completion signal infinitely. Consequently, the test will be blocked until it is killed. To fix that problem, StepVerifier offers a cancellation API that unsubscribes from the source when some expectations are satisfied, as shown in the following code:

Flux<String> websocketPublisher = ...
StepVerifier
   .create(websocketPublisher)
   .expectSubscription()
   .expectNext("Connected")
   .expectNext("Price: $12.00")
   .thenCancel()
   .verify();

The preceding code tells us that we will disconnect or unsubscribe from the WebSocket after getting Connected and after receiving the Price: $12.00 message.

Another crucial stage during the system's verification is checking for the backpressure behavior of the Publisher. For example, interaction with an external system over WebSocket results in a push-only Publisher. A naive way to prevent such behavior is by protecting downstream with the .onBackpressureBuffer() operator. To check that the system behaves as expected with the selected backpressure strategy, we have to control the subscriber demand manually. In order to do this, StepVerifier offers the .thenRequest() method, which allows us to control the subscriber demand. This is depicted in the following code:

Flux<String> websocketPublisher = ...
Class<Exception> expectedErrorClass = 
    reactor.core.Exceptions.failWithOverflow().getClass();

StepVerifier
   .create(websocketPublisher.onBackpressureBuffer(5), 0)
   .expectSubscription()
   .thenRequest(1)
   .expectNext("Connected")
   .thenRequest(1)
   .expectNext("Price: $12.00")
   .expectError(expectedErrorClass)
   .verify();

From the preceding example, we might learn how to use the .thenRequest() method for verifying backpressure behavior. In turn, it is expected that at some point in time, the overflow will take place and we will get an overflow error. Please note that in the preceding example, we used the StepVerifier.create() method's overload, which accepts the initial subscriber's demand as the second argument. In one argument method's overload, the default demand is Long.MAX_VALUE, which refers to unlimited demand.

One of the advanced features offered by the StepVerifiers API is the ability to run an additional action after a particular verification. For example, in the case where elements producing the process require some additional external interaction, this might be done with the .then() method.

We will also use TestPublisher from the test package of the Reactor Core library. TestPublisher implements the Reactive Stream Publisher and makes it possible to directly trigger the onNext(), onComplete(), and onError() events for testing purposes. The next example demonstrates how to trigger new events during the test execution:

TestPublisher<String> idsPublisher = TestPublisher.create();

StepVerifier
   .create(walletsRepository.findAllById(idsPublisher))
   .expectSubscription()
   .then(() -> idsPublisher.next("1"))                               // (1)
   .assertNext(w -> assertThat(w, hasProperty("id", equalTo("1"))))  // (2)
   .then(() -> idsPublisher.next("2"))                               // (3) 
   .assertNext(w -> assertThat(w, hasProperty("id", equalTo("2"))))  // (4)
   .then(idsPublisher::complete)                                     // (5)
   .expectComplete()
   .verify();

In this example, it is a requirement to verify that WalletRepository searching wallets are correctly given ids. In turn, one of the specific requirements of the wallet repository is searching for data as it comes, which means that the upstream should be a hot Publisher. In our example, we use TestPublisher.next() in combination with StepVerifier.then(). Steps (1) and (3) send new requests only after previous steps were verified. Step (2) verifies that requests generated with step (1) were processed successfully and correspondingly, while line (4) verifies step (3). Step (5) commands TestPublisher to complete the request stream, after which StepVerifier verifies that the response stream is also completed.

This technique plays an important role by enabling event production after the subscription has actually happened. In this way, we may verify that emitted IDs have been found right after that action, and that the walletsRepository behaves as expected.