Testing Reactive Streams Using StepVerifier and TestPublisher – 使用StepVerifier和TestPublisher测试反应式数据流

最后修改: 2018年 12月 21日

中文/混合/英文(键盘快捷键：t)

1. Overview

1.概述

In this tutorial, we’ll take a close look at testing reactive streams with StepVerifier and TestPublisher.

在本教程中，我们将仔细研究用StepVerifier和TestPublisher来测试reactive streams。

We’ll base our investigation on a Spring Reactor application containing a chain of reactor operations.

我们将基于一个Spring Reactor应用程序进行调查，该应用程序包含一个反应器操作链。

2. Maven Dependencies

2.Maven的依赖性

Spring Reactor comes with several classes for testing reactive streams.

Spring Reactor带有几个用于测试反应式流的类。

We can get these by adding the reactor-test dependency:

我们可以通过添加reactor-test依赖项来获得这些。

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <scope>test</scope>
    <version>3.2.3.RELEASE</version>
</dependency>

3. StepVerifier

3.步骤验证器

In general, reactor-test has two main uses:

一般来说，reactor-test有两个主要用途。

  • creating a step-by-step test with StepVerifier
  • producing predefined data with TestPublisher to test downstream operators

The most common case in testing reactive streams is when we have a publisher (a Flux or Mono) defined in our code. We want to know how it behaves when someone subscribes. 

在测试反应式流时，最常见的情况是我们在代码中定义了一个发布者（一个Flux或Mono）。我们想知道当有人订阅时，它是如何表现的。

With the StepVerifier API, we can define our expectations of published elements in terms of what elements we expect and what happens when our stream completes.

通过StepVerifier API，我们可以在我们期望的元素和我们的流完成时发生的情况方面定义我们对发布的元素的期望。

First of all, let’s create a publisher with some operators.

首先，让我们创建一个带有一些运算符的发布器。

We’ll use a Flux.just(T elements). This method will create a Flux that emits given elements and then completes.

我们将使用一个Flux.just(T elements).这个方法将创建一个Flux，它发射给定的元素，然后完成。

Since advanced operators are beyond the scope of this article, we’ll just create a simple publisher that outputs only four-letter names mapped to uppercase:

由于高级运算符超出了本文的范围，我们只需创建一个简单的发布器，只输出映射为大写的四个字母的名字。

Flux<String> source = Flux.just("John", "Monica", "Mark", "Cloe", "Frank", "Casper", "Olivia", "Emily", "Cate")
  .filter(name -> name.length() == 4)
  .map(String::toUpperCase);

3.1. Step-By-Step Scenario

3.1. 分步方案

Now, let’s test our source with StepVerifier in order to test what will happen when someone subscribes:

现在，让我们用StepVerifier 测试我们的source ，以测试当有人订阅时将会发生什么。

StepVerifier
  .create(source)
  .expectNext("JOHN")
  .expectNextMatches(name -> name.startsWith("MA"))
  .expectNext("CLOE", "CATE")
  .expectComplete()
  .verify();

First, we create a StepVerifier builder with the create method.

首先，我们用create方法创建一个StepVerifierbuilder。

Next, we wrap our Flux source, which is under test. The first signal is verified with expectNext(T element), but really, we can pass any number of elements to expectNext.

接下来，我们将我们的Flux源包裹起来，它正在接受测试。第一个信号是通过expectNext(T element)验证的，但实际上，我们可以向expectNext传递任何数量的元素。。

We can also use expectNextMatches and provide a Predicate<T> for a more custom match.

我们也可以使用expectNextMatches ，并提供一个Predicate<T>以获得更多的自定义匹配。

For our last expectation, we expect that our stream completes.

对于我们的最后一个期望，我们期望我们的流完成了。

And finally, we use verify() to trigger our test.

最后，我们使用verify()来触发我们的测试。

3.2. Exceptions in StepVerifier

3.2.StepVerifier中的异常情况

Now, let’s concatenate our Flux publisher with Mono.

现在，让我们把我们的Flux发布者与Mono.连接起来。

We’ll have this Mono terminate immediately with an error when subscribed to:

我们会让这个Mono在订阅时立即以错误结束。

Flux<String> error = source.concatWith(
  Mono.error(new IllegalArgumentException("Our message"))
);

Now, after four all elements, we expect our stream to terminate with an exception:

现在，在四个元素之后，我们期望我们的流以一个异常终止。

StepVerifier
  .create(error)
  .expectNextCount(4)
  .expectErrorMatches(throwable -> throwable instanceof IllegalArgumentException &&
    throwable.getMessage().equals("Our message")
  ).verify();

We can use only one method to verify exceptions. The OnError signal notifies the subscriber that the publisher is closed with an error state. Therefore, we can’t add more expectations afterward.

我们只能使用一种方法来验证异常。OnError信号通知订阅者，发布者以错误状态关闭。因此，我们不能在之后添加更多的期望。

If it’s not necessary to check the type and message of the exception at once, then we can use one of the dedicated methods:

如果没有必要一次性检查异常的类型和消息，那么我们可以使用其中一个专用方法。

  • expectError() – expect any kind of error
  • expectError(Class<? extends Throwable> clazz) – expect an error of a specific type
  • expectErrorMessage(String errorMessage) – expect an error having a specific message
  • expectErrorMatches(Predicate<Throwable> predicate) – expect an error that matches a given predicate
  • expectErrorSatisfies(Consumer<Throwable> assertionConsumer) – consume a Throwable in order to do a custom assertion

3.3. Testing Time-Based Publishers

3.3.测试基于时间的发布者

Sometimes our publishers are time-based.

有时我们的出版商是基于时间的。。

For example, suppose that in our real-life application, we have a one-day delay between events. Now, obviously, we don’t want our tests to run for an entire day to verify expected behavior with such a delay.

例如，假设在我们的实际应用中，我们在事件之间有一天的延迟。现在，很明显，我们不希望我们的测试运行一整天来验证这种延迟的预期行为。

StepVerifier.withVirtualTime builder is designed to avoid long-running tests.

StepVerifier.withVirtualTime构建器旨在避免长时间运行的测试。

We create a builder by calling withVirtualTime. Note that this method doesn’t take Flux as input. Instead, it takes a Supplier, which lazily creates an instance of the tested Flux after having the scheduler set up.

我们通过调用withVirtualTime来创建一个构建器。 注意这个方法不接受Flux作为输入。相反，它接受一个Supplier，在设置了调度器之后，它懒散地创建一个被测试的Flux实例。

To demonstrate how we can test for an expected delay between events, let’s create a Flux with an interval of one second that runs for two seconds. If the timer runs correctly, we should only get two elements:

为了演示我们如何测试事件之间的预期延迟，让我们创建一个Flux，其间隔时间为一秒，运行两秒。如果定时器正确运行，我们应该只得到两个元素：。

StepVerifier
  .withVirtualTime(() -> Flux.interval(Duration.ofSeconds(1)).take(2))
  .expectSubscription()
  .expectNoEvent(Duration.ofSeconds(1))
  .expectNext(0L)
  .thenAwait(Duration.ofSeconds(1))
  .expectNext(1L)
  .verifyComplete();

Note that we should avoid instantiating the Flux earlier in the code and then having the Supplier returning this variable. Instead, we should always instantiate Flux inside the lambda.

请注意，我们应该避免在代码中提前实例化Flux，然后让Supplier返回这个变量。相反，我们应该始终在lambda中实例化Flux 。

There are two major expectation methods that deal with time:

有两种处理时间的主要预期方法。

  • thenAwait(Duration duration) – pauses the evaluation of the steps; new events may occur during this time
  • expectNoEvent(Duration duration) – fails when any event appears during the duration; the sequence will pass with a given duration

Please notice that the first signal is the subscription event, so every expectNoEvent(Duration duration) should be preceded with expectSubscription().

请注意，第一个信号是订阅事件，所以每一个expectNoEvent(Duration duration)前面都应该有expectSubscription()。

3.4. Post-Execution Assertions with StepVerifier

3.4.使用StepVerifier的执行后断言

So, as we’ve seen, it’s straightforward to describe our expectations step-by-step.

因此，正如我们所看到的，一步步地描述我们的期望是很直接的。

However, sometimes we need to verify additional state after our whole scenario played out successfully.

然而，有时我们需要在整个场景成功播放后验证其他状态。

Let’s create a custom publisher. It will emit a few elements, then complete, pause, and emit one more element, which we’ll drop:

让我们创建一个自定义发布器。它将发射几个元素，然后完成，暂停，再发射一个元素，我们将放弃它。

Flux<Integer> source = Flux.<Integer>create(emitter -> {
    emitter.next(1);
    emitter.next(2);
    emitter.next(3);
    emitter.complete();
    try {
        Thread.sleep(1000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    emitter.next(4);
}).filter(number -> number % 2 == 0);

We expect that it will emit a 2, but drop a 4, since we called emitter.complete first.

我们预计它将发射一个2，但放弃一个4，因为我们先调用了emitter.complete。。

So, let’s verify this behavior by using verifyThenAssertThat. This method returns StepVerifier.Assertions on which we can add our assertions:

因此，让我们通过使用verifyThenAssertThat来验证这一行为。该方法返回StepVerifier.Assertions，我们可以在上面添加我们的断言。

@Test
public void droppedElements() {
    StepVerifier.create(source)
      .expectNext(2)
      .expectComplete()
      .verifyThenAssertThat()
      .hasDropped(4)
      .tookLessThan(Duration.ofMillis(1050));
}

4. Producing Data with TestPublisher

4.用TestPublisher制作数据

Sometimes, we might need some special data in order to trigger the chosen signals.

有时，我们可能需要一些特殊的数据，以触发所选择的信号。

For instance, we may have a very particular situation that we want to test.

例如，我们可能有一个非常特殊的情况，我们想测试一下。

Alternatively, we may choose to implement our own operator and want to test how it behaves.

另外，我们可能会选择实现我们自己的操作者，并想测试它的行为方式。

For both cases, we can use TestPublisher<T>, which allows us to programmatically trigger miscellaneous signals:

对于这两种情况，我们可以使用TestPublisher<T>，它允许我们以编程方式触发各种信号：。

  • next(T value) or next(T value, T rest) – send one or more signals to subscribers
  • emit(T value) – same as next(T) but invokes complete() afterwards
  • complete() – terminates a source with the complete signal
  • error(Throwable tr) – terminates a source with an error
  • flux() – convenient method to wrap a TestPublisher into Flux
  • mono() – same us flux() but wraps to a Mono

4.1. Creating a TestPublisher

4.1.创建一个TestPublisher

Let’s create a simple TestPublisher that emits a few signals and then terminates with an exception:

让我们创建一个简单的TestPublisher，它发出一些信号，然后用一个异常终止。

TestPublisher
  .<String>create()
  .next("First", "Second", "Third")
  .error(new RuntimeException("Message"));

4.2. TestPublisher in Action

4.2.TestPublisher运行中

As we mentioned earlier, we may sometimes want to trigger a finely chosen signal that closely matches to a particular situation.

正如我们前面提到的，我们有时可能希望触发一个精心选择的、与特定情况密切相关的信号。

Now, it’s especially important in this case that we have complete mastery over the source of the data. To achieve this, we can again rely on TestPublisher.

现在，在这种情况下，我们完全掌握了数据的来源，这一点特别重要。为了实现这一点，我们可以再次依靠TestPublisher。

First, let’s create a class that uses Flux<String> as the constructor parameter to perform the operation getUpperCase():

首先，让我们创建一个使用Flux<String> 作为构造参数的类来执行getUpperCase()操作。

class UppercaseConverter {
    private final Flux<String> source;

    UppercaseConverter(Flux<String> source) {
        this.source = source;
    }

    Flux<String> getUpperCase() {
        return source
          .map(String::toUpperCase);
    }   
}

Suppose that UppercaseConverter is our class with complex logic and operators, and we need to supply very particular data from the source publisher.

假设UppercaseConverter是我们的类，具有复杂的逻辑和运算符，而且我们需要从sourcepublisher提供非常特别的数据。

We can easily achieve this with TestPublisher:

我们可以用TestPublisher轻松实现这一目标：。

final TestPublisher<String> testPublisher = TestPublisher.create();

UppercaseConverter uppercaseConverter = new UppercaseConverter(testPublisher.flux());

StepVerifier.create(uppercaseConverter.getUpperCase())
  .then(() -> testPublisher.emit("aA", "bb", "ccc"))
  .expectNext("AA", "BB", "CCC")
  .verifyComplete();

In this example, we create a test Flux publisher in the UppercaseConverter constructor parameter. Then, our TestPublisher emits three elements and completes.

在这个例子中，我们在UppercaseConverterconstructor参数中创建了一个测试Fluxpublisher。然后，我们的TestPublisher发射了三个元素并完成了。

4.3. Misbehaving TestPublisher

4.3.行为不端的TestPublisher

On the other hand, we can create a misbehaving TestPublisher with the createNonCompliant factory method. We need to pass in the constructor one enum value from TestPublisher.Violation. These values specify which parts of specifications our publisher may overlook.

另一方面，我们可以用createNonCompliant factory方法创建一个行为不端的TestPublisher。我们需要在构造函数中传递一个来自TestPublisher.violation.的枚举值，这些值指定我们的发布者可能忽略规范的哪一部分。

Let’s take a look at a TestPublisher that won’t throw a NullPointerException for the null element:

让我们来看看一个TestPublisher，它不会为null元素抛出一个NullPointerException。

TestPublisher
  .createNoncompliant(TestPublisher.Violation.ALLOW_NULL)
  .emit("1", "2", null, "3");

In addition to ALLOW_NULL, we can also use TestPublisher.Violation to:

除了ALLOW_NULL，我们还可以使用TestPublisher.Violation来。

  • REQUEST_OVERFLOW – allows calling next() without throwing an IllegalStateException when there’s an insufficient number of requests
  • CLEANUP_ON_TERMINATE – allows sending any termination signal several times in a row
  • DEFER_CANCELLATION – allows us to ignore cancellation signals and continue with emitting elements

5. Conclusion

5.总结

In this article, we discussed various ways of testing reactive streams from the Spring Reactor project.

在这篇文章中，我们讨论了从Spring Reactor项目中测试反应式流的各种方法。。

First, we saw how to use StepVerifier to test publishers. Then, we saw how to use TestPublisher. Similarly, we saw how to operate with a misbehaving TestPublisher.

首先，我们看到如何使用StepVerifier来测试发布者。然后，我们看到了如何使用TestPublisher。同样地，我们看到了如何操作一个不正常的TestPublisher。

As usual, the implementation of all our examples can be found in the Github project.

像往常一样，我们所有例子的实现都可以在Github项目中找到。