@Test public void subscribeWithConsumerAndErrorConsumer() { Flux<Integer> flux = workshop.subscribeWithConsumerAndErrorConsumer(); StepVerifier.create(flux) .expectNext(1, 2, 3).verifyError(); }
public Flux<Integer> subscribeWithConsumerAndErrorConsumer() { Flux<Integer> flux = Flux.just(1, 2, 3, 4).map(i -> { if (i != 4) { return i; } else { throw new IllegalStateException("error"); } }); flux.log().subscribe(integer -> log.info("{}", integer), throwable -> log.info("{}", throwable.getMessage())); return flux; }
Part05Subscribe { public Flux<Integer> subscribeWithConsumerAndErrorConsumer() { Flux<Integer> flux = Flux.just(1, 2, 3, 4).map(i -> { if (i != 4) { return i; } else { throw new IllegalStateException("error"); } }); flux.log().subscribe(integer -> log.info("{}", integer), throwable -> log.info("{}", throwable.getMessage())); return flux; } }
Part05Subscribe { public Flux<Integer> subscribeWithConsumerAndErrorConsumer() { Flux<Integer> flux = Flux.just(1, 2, 3, 4).map(i -> { if (i != 4) { return i; } else { throw new IllegalStateException("error"); } }); flux.log().subscribe(integer -> log.info("{}", integer), throwable -> log.info("{}", throwable.getMessage())); return flux; }  }
Part05Subscribe { public Flux<Integer> subscribeWithConsumerAndErrorConsumer() { Flux<Integer> flux = Flux.just(1, 2, 3, 4).map(i -> { if (i != 4) { return i; } else { throw new IllegalStateException("error"); } }); flux.log().subscribe(integer -> log.info("{}", integer), throwable -> log.info("{}", throwable.getMessage())); return flux; }  Flux<Integer> subscribeEmpty(); Flux<Integer> subscribeWithConsumer(); Flux<Integer> subscribeWithConsumerAndCompleteConsumer(); Flux<Integer> subscribeWithConsumerAndErrorConsumer(); Flux<Integer> subscribeWithSubscriptionConsumer(); Flux<Integer> subscribeWithSubscription(); }
Part05Subscribe { public Flux<Integer> subscribeWithConsumerAndErrorConsumer() { Flux<Integer> flux = Flux.just(1, 2, 3, 4).map(i -> { if (i != 4) { return i; } else { throw new IllegalStateException("error"); } }); flux.log().subscribe(integer -> log.info("{}", integer), throwable -> log.info("{}", throwable.getMessage())); return flux; }  Flux<Integer> subscribeEmpty(); Flux<Integer> subscribeWithConsumer(); Flux<Integer> subscribeWithConsumerAndCompleteConsumer(); Flux<Integer> subscribeWithConsumerAndErrorConsumer(); Flux<Integer> subscribeWithSubscriptionConsumer(); Flux<Integer> subscribeWithSubscription();  }