@Test @Ignore public void testBackpressureWorking() { EventRepository<String> repository = mock(EventRepository.class); EventSubscriptionPositionSource positionSource = mock(EventSubscriptionPositionSource.class); WorkQueueProcessor<EventSubscriptionUpdate<String>> processor = WorkQueueProcessor.create(); Flux<EventSubscriptionUpdate<String>> eventSource = Flux .fromStream(IntStream .range(0, 1000000) .mapToObj(this::wrapIntAsEvent) .map(EventSubscriptionUpdate::ofEvent)) .doOnNext(e -> { lastProducedValue = e.getEvent().getEvent(); }); eventSource.subscribe(processor); when(repository.getPublisher(any())).thenReturn(processor); 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(5000); token.stop(); logger.info( "Last produced value was {}, last seen {}, total events seen: {}", lastProducedValue, subscriptionHandler.getLastSeenValue(), subscriptionHandler.getTotalEvents()); }
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();  }