CancellationPropagationIntegrationTest.java 文件源码

@Test
public void clientCanCancelServerStreamImplicitly() throws InterruptedException {
    ReactorNumbersGrpc.ReactorNumbersStub stub = ReactorNumbersGrpc.newReactorStub(channel);
    Flux<NumberProto.Number> test = stub
            .responsePressure(Mono.just(Empty.getDefaultInstance()))
            .doOnNext(number -> System.out.println(number.getNumber(0)))
            .doOnError(throwable -> System.out.println(throwable.getMessage()))
            .doOnComplete(() -> System.out.println("Completed"))
            .doOnCancel(() -> System.out.println("Client canceled"))
            .take(10);

    Disposable subscription = test.publish().connect();

    Thread.sleep(1000);

    assertThat(svc.wasCanceled()).isTrue();
}