CancellationPropagationIntegrationTest.java 文件源码

@Override
public Flux<NumberProto.Number> responsePressure(Mono<Empty> request) {
    // Produce a very long sequence
    return Flux
            .fromIterable(IntStream.range(0, NUMBER_OF_STREAM_ELEMENTS)::iterator)
            .delayElements(Duration.ofMillis(SEQUENCE_DELAY_MILLIS))
            .doOnNext(i -> lastNumberProduced.set(i))
            .map(CancellationPropagationIntegrationTest::protoNum)
            .doOnCancel(() -> {
                wasCanceled.set(true);
                System.out.println("Server canceled");
            });
}