CancellationPropagationIntegrationTest.java 文件源码

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