### Question:
AbstractEventHandler { public void start() { this.scheduler = this.createScheduler(); this.subscription = Flux.from(this.publisher).subscribeOn(this.scheduler).log(this.log.getName(), Level.FINEST) .doOnSubscribe((s) -> this.log.debug("Subscribed to {} events", this.eventType)).ofType(this.eventType) .cast(this.eventType).transform(this::handle).retryWhen(Retry.any().retryMax(Long.MAX_VALUE) .doOnRetry((ctx) -> this.log.warn("Unexpected error", ctx.exception()))) .subscribe(); } protected  AbstractEventHandler(Publisher<InstanceEvent> publisher, Class<T> eventType); void start(); void stop();  }### Answer:
@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(firstEvent, errorEvent, secondEvent)).expectNext(firstEvent, secondEvent) .thenCancel().verify(Duration.ofSeconds(1)); }
@Test public void should_filter() { TestPublisher<InstanceEvent> testPublisher = TestPublisher.create(); TestEventHandler eventHandler = new TestEventHandler(testPublisher.flux()); eventHandler.start(); StepVerifier.create(eventHandler.getFlux()).expectSubscription() .then(() -> testPublisher.next(firstEvent, ignoredEvent, secondEvent)) .expectNext(firstEvent, secondEvent).thenCancel().verify(Duration.ofSeconds(1)); }