@Test public void subscribeWithSubscription() { Flux<Integer> flux = workshop.subscribeWithSubscription(); StepVerifier.create(flux) .expectNext(1, 2, 3).verifyComplete(); }
public Flux<Integer> subscribeWithSubscription() { Flux<Integer> flux = Flux.just(1, 2, 3); flux.log().subscribe(new Subscriber<Integer>() { Subscription s; @Override public void onSubscribe(Subscription s) { this.s = s; s.request(1); } @Override public void onNext(Integer integer) { s.request(integer); } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); return flux; }
Part05Subscribe { public Flux<Integer> subscribeWithSubscription() { Flux<Integer> flux = Flux.just(1, 2, 3); flux.log().subscribe(new Subscriber<Integer>() { Subscription s; @Override public void onSubscribe(Subscription s) { this.s = s; s.request(1); } @Override public void onNext(Integer integer) { s.request(integer); } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); return flux; } }
Part05Subscribe { public Flux<Integer> subscribeWithSubscription() { Flux<Integer> flux = Flux.just(1, 2, 3); flux.log().subscribe(new Subscriber<Integer>() { Subscription s; @Override public void onSubscribe(Subscription s) { this.s = s; s.request(1); } @Override public void onNext(Integer integer) { s.request(integer); } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); return flux; }  }
Part05Subscribe { public Flux<Integer> subscribeWithSubscription() { Flux<Integer> flux = Flux.just(1, 2, 3); flux.log().subscribe(new Subscriber<Integer>() { Subscription s; @Override public void onSubscribe(Subscription s) { this.s = s; s.request(1); } @Override public void onNext(Integer integer) { s.request(integer); } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); return flux; }  Flux<Integer> subscribeEmpty(); Flux<Integer> subscribeWithConsumer(); Flux<Integer> subscribeWithConsumerAndCompleteConsumer(); Flux<Integer> subscribeWithConsumerAndErrorConsumer(); Flux<Integer> subscribeWithSubscriptionConsumer(); Flux<Integer> subscribeWithSubscription(); }
Part05Subscribe { public Flux<Integer> subscribeWithSubscription() { Flux<Integer> flux = Flux.just(1, 2, 3); flux.log().subscribe(new Subscriber<Integer>() { Subscription s; @Override public void onSubscribe(Subscription s) { this.s = s; s.request(1); } @Override public void onNext(Integer integer) { s.request(integer); } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); return flux; }  Flux<Integer> subscribeEmpty(); Flux<Integer> subscribeWithConsumer(); Flux<Integer> subscribeWithConsumerAndCompleteConsumer(); Flux<Integer> subscribeWithConsumerAndErrorConsumer(); Flux<Integer> subscribeWithSubscriptionConsumer(); Flux<Integer> subscribeWithSubscription();  }