使用schedulers.single（）在Reactor步进验证器中重新抛出AssertionErrors

Question

我们使用Project Reactor按照下面ServiceTest中的代码异步运行特定操作。为了测试这段代码，按照下面的ServiceTest，在为异步操作设置Mono时，我们使用doonnext将Mono的结果传递给DirectProcessor，然后使用stepVerifier执行测试调用和断言。

stepverifier#assertNext的JavaDoc读取

使用者抛出的任何AssertionErrors都将在验证期间重新抛出。

是否可以，如果可以，如何使用单个调度程序，并在验证过程中根据Javadoc重新抛出AssertionErrors？

@Service
@RequiredArgsConstructor
public class Service implements WithReactive, WithTestProcessor<Response> {

    @Getter
    @Setter
    private DirectProcessor<Response> processor = DirectProcessor.create();

    @Setter
    private Scheduler scheduler = Schedulers.single();

    public void doAction() {
        Mono.fromSupplier(this::doActionAsync)
            .doOnNext(processor::onNext)
            .subscribeOn(scheduler)
            .subscribe();
    }

    private Response doActionAsync() {
        ...
    }

    ...

}

public interface WithReactive {
    void setScheduler(Scheduler scheduler);
}

public interface WithTestProcessor<T> {
    void setProcessor(DirectProcessor<T> processor);
    DirectProcessor<T> getProcessor();
}

@RunWith(SpringRunner.class)
@SpringBootTest
public class ServiceTest {

    @Inject
    private Collection<WithTestProcessor> withTestProcessors;

    @Before
    public void setTestProcessors() {
        withTestProcessors.forEach(withTestProcessor -> withTestProcessor.setProcessor(DirectProcessor.create()));
    }

    @Inject
    private Collection<WithReactive> withReactives;

    @Before
    public void makeReactiveSynchronous() {
        withReactives.forEach(withReactive -> withReactive.setScheduler(Schedulers.immediate()));
    }

    @Test
    private void test() {
      StepVerifier.create(service.getProcessor())
          .then(service::doAction)
          .assertNext(response -> assertThat(logExtractor.getInsertsByTable("assets")).hasSize(1))
          .thenCancel()
          .verify();
    }
}

Answer 1

这是三个因素的结合：最初的然后，订阅与验证同时发生的事实，由于subscribeon，以及thencancel。

一种解决方法是，在StepVerifier执行thencancel之前，给thenawait(Duration.ofMillis(10))足够的时间让onnext发生，方法是在thencancel之前放一个thenawait(duration.ofMillis(10))。