@Test public void should_resubscribe_after_error() { TestPublisher<InstanceEvent> testPublisher = TestPublisher.create(); TestEventHandler eventHandler = new TestEventHandler(testPublisher.flux()); eventHandler.start(); StepVerifier.create(eventHandler.getFlux()) .expectSubscription() .then(() -> testPublisher.next(event)) .expectNext(event) .then(() -> testPublisher.next(errorEvent)) .expectNoEvent(Duration.ofMillis(100L)) .then(() -> testPublisher.next(event)) .expectNext(event) .thenCancel() .verify(Duration.ofSeconds(5)); }
public void start() { subscription = Flux.from(publisher) .log(log.getName(), Level.FINEST) .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType)) .ofType(eventType) .cast(eventType) .transform(this::handle) .retryWhen(Retry.any() .retryMax(Long.MAX_VALUE) .doOnRetry(ctx -> log.warn("Unexpected error", ctx.exception()))) .subscribe(); }
AbstractEventHandler { public void start() { subscription = Flux.from(publisher) .log(log.getName(), Level.FINEST) .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType)) .ofType(eventType) .cast(eventType) .transform(this::handle) .retryWhen(Retry.any() .retryMax(Long.MAX_VALUE) .doOnRetry(ctx -> log.warn("Unexpected error", ctx.exception()))) .subscribe(); } }
AbstractEventHandler { public void start() { subscription = Flux.from(publisher) .log(log.getName(), Level.FINEST) .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType)) .ofType(eventType) .cast(eventType) .transform(this::handle) .retryWhen(Retry.any() .retryMax(Long.MAX_VALUE) .doOnRetry(ctx -> log.warn("Unexpected error", ctx.exception()))) .subscribe(); } protected  AbstractEventHandler(Publisher<InstanceEvent> publisher, Class<T> eventType); }
AbstractEventHandler { public void start() { subscription = Flux.from(publisher) .log(log.getName(), Level.FINEST) .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType)) .ofType(eventType) .cast(eventType) .transform(this::handle) .retryWhen(Retry.any() .retryMax(Long.MAX_VALUE) .doOnRetry(ctx -> log.warn("Unexpected error", ctx.exception()))) .subscribe(); } protected  AbstractEventHandler(Publisher<InstanceEvent> publisher, Class<T> eventType); void start(); void stop(); }
AbstractEventHandler { public void start() { subscription = Flux.from(publisher) .log(log.getName(), Level.FINEST) .doOnSubscribe(s -> log.debug("Subscribed to {} events", eventType)) .ofType(eventType) .cast(eventType) .transform(this::handle) .retryWhen(Retry.any() .retryMax(Long.MAX_VALUE) .doOnRetry(ctx -> log.warn("Unexpected error", ctx.exception()))) .subscribe(); } protected  AbstractEventHandler(Publisher<InstanceEvent> publisher, Class<T> eventType); void start(); void stop();  }