@Test @Ignore public void testRetriesOnStreamError() { EventRepository<String> repository = mock(EventRepository.class); EventSubscriptionPositionSource positionSource = mock(EventSubscriptionPositionSource.class); when(repository.getPublisher(any())).then(position -> { WorkQueueProcessor<EventRecord<String>> processor = WorkQueueProcessor.create(); Flux<EventRecord<String>> eventSource = Flux .fromStream(IntStream.range(0, 1000000).mapToObj(this::wrapIntAsEvent)) .doOnNext(e -> { lastProducedValue = e.getEvent(); }); eventSource.subscribe(processor); return processor.take(100).thenMany(Flux.error(new RuntimeException("fail!"))); }); when(positionSource.getSubscriptionPosition()).thenReturn(null); SlowSubscriptionHandler<String> subscriptionHandler = new SlowSubscriptionHandler<>(); EventSubscriptionManager subscriptionManager = new EventSubscriptionManager<>( repository, positionSource, subscriptionHandler, new SubscriptionWorkerConfig().withBatchSize(64)); SubscriptionToken token = subscriptionManager.start(); sleep(100000); token.stop(); }
public SubscriptionToken start() { workerThread.start(); return subscriptionWorker; }
EventSubscriptionManager { public SubscriptionToken start() { workerThread.start(); return subscriptionWorker; } }
EventSubscriptionManager { public SubscriptionToken start() { workerThread.start(); return subscriptionWorker; } EventSubscriptionManager(
            final EventRepository<T> repository,
            final EventSubscriptionPositionSource positionSource,
            final EventSubscriptionHandler<T> subscriptionHandler,
            final SubscriptionWorkerConfig config); }
EventSubscriptionManager { public SubscriptionToken start() { workerThread.start(); return subscriptionWorker; } EventSubscriptionManager(
            final EventRepository<T> repository,
            final EventSubscriptionPositionSource positionSource,
            final EventSubscriptionHandler<T> subscriptionHandler,
            final SubscriptionWorkerConfig config); SubscriptionToken start(); }
EventSubscriptionManager { public SubscriptionToken start() { workerThread.start(); return subscriptionWorker; } EventSubscriptionManager(
            final EventRepository<T> repository,
            final EventSubscriptionPositionSource positionSource,
            final EventSubscriptionHandler<T> subscriptionHandler,
            final SubscriptionWorkerConfig config); SubscriptionToken start();  }