使用Schedulers.single（）在Reactor StepVerifier中反悔断言错误(Rethrow AssertionErrors in Reactor StepVerifier us

我们使用Project Reactor按照下面的ServiceTest中的代码异步运行特定的操作。 为了测试这段代码，根据下面的ServiceTest ，当为异步操作设置Mono时 ，我们让Mono将其结果传递给具有测试有权访问的doOnNext的DirectProcessor，然后使用StepVerifier执行我们的测试调用和断言。

StepVerifier＃assertNext的JavaDoc读取

验证期间，消费者抛出的任何AssertionErrors都将被重新抛出。

我们发现只有在使用立即调度程序（ Schedulers.immediate（） ）时才会出现这种情况，并且在使用单个调度程序（ Schedulers.single（） ）时不成立。 当使用单个调度程序时，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();
    }
}
We are using Project Reactor  to run a particular operation asynchronously as per the code in ServiceTest below. To test this code, as per ServiceTest below, when setting up the Mono for the async operation we make the Mono pass it's result to a DirectProcessor with doOnNext that the test has access to, and then carry out our test call and assertions with StepVerifier.
 
The JavaDoc of StepVerifier#assertNext reads
 
 
 
Any AssertionErrors thrown by the consumer will be rethrown during verification.
 
 
We have found that is true only when the immediate scheduler (Schedulers.immediate()) is used and is not true when the single scheduler (Schedulers.single()) is used. When the single scheduler is used, AssertionErrors are not re-thrown, i.e. the test always passes.
 
Is it possible, and if so, how, to use the single scheduler and have AssertionErrors rethrown during verification as per the JavaDoc?
 
@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();
    }
}

                
最满意答案
                
                    
                        
 这是三个因素的组合：首先，由于subscribeOn和thenCancel ， subscribeOn与验证并行发生。 
 
 一种解决方法是通过在onNext之前放置thenAwait(Duration.ofMillis(10)) ，在StepVerifier执行thenCancel之前给onNext足够的时间。 
This is a combination of three factors: the initial then, the fact that subscription happens in parallel of the verification due to subscribeOn and the thenCancel.
 
One workaround is to give enough time to the onNext to happen before the StepVerifier executes thenCancel, by putting a thenAwait(Duration.ofMillis(10)) before the thenCancel.