@Test @Timeout(value = 5) void testConsumerStopsTailingWhenItFallsBehindTheCache() throws Exception { newTestSubject(CACHED_EVENTS, FETCH_DELAY, 20, OPTIMIZE_EVENT_CONSUMPTION); TrackingEventStream stream = testSubject.openStream(null); assertFalse(stream.hasNextAvailable()); testSubject.publish(createEvents(CACHED_EVENTS)); Thread.sleep(100); reset(storageEngine); assertTrue(stream.hasNextAvailable()); TrackedEventMessage<?> firstEvent = stream.nextAvailable(); verifyZeroInteractions(storageEngine); testSubject.publish(createEvent(CACHED_EVENTS), createEvent(CACHED_EVENTS + 1)); Thread.sleep(100); reset(storageEngine); assertTrue(stream.hasNextAvailable()); verify(storageEngine).readEvents(firstEvent.trackingToken(), false); }
@Override public TrackingEventStream openStream(TrackingToken trackingToken) { Node node = findNode(trackingToken); EventConsumer eventConsumer; if (node != null && optimizeEventConsumption) { eventConsumer = new EventConsumer(node); tailingConsumers.add(eventConsumer); } else { eventConsumer = new EventConsumer(trackingToken); } return eventConsumer; }
EmbeddedEventStore extends AbstractEventStore { @Override public TrackingEventStream openStream(TrackingToken trackingToken) { Node node = findNode(trackingToken); EventConsumer eventConsumer; if (node != null && optimizeEventConsumption) { eventConsumer = new EventConsumer(node); tailingConsumers.add(eventConsumer); } else { eventConsumer = new EventConsumer(trackingToken); } return eventConsumer; } }
EmbeddedEventStore extends AbstractEventStore { @Override public TrackingEventStream openStream(TrackingToken trackingToken) { Node node = findNode(trackingToken); EventConsumer eventConsumer; if (node != null && optimizeEventConsumption) { eventConsumer = new EventConsumer(node); tailingConsumers.add(eventConsumer); } else { eventConsumer = new EventConsumer(trackingToken); } return eventConsumer; } protected  EmbeddedEventStore(Builder builder); }
EmbeddedEventStore extends AbstractEventStore { @Override public TrackingEventStream openStream(TrackingToken trackingToken) { Node node = findNode(trackingToken); EventConsumer eventConsumer; if (node != null && optimizeEventConsumption) { eventConsumer = new EventConsumer(node); tailingConsumers.add(eventConsumer); } else { eventConsumer = new EventConsumer(trackingToken); } return eventConsumer; } protected  EmbeddedEventStore(Builder builder); static Builder builder(); @PreDestroy void shutDown(); @Override TrackingEventStream openStream(TrackingToken trackingToken); }
EmbeddedEventStore extends AbstractEventStore { @Override public TrackingEventStream openStream(TrackingToken trackingToken) { Node node = findNode(trackingToken); EventConsumer eventConsumer; if (node != null && optimizeEventConsumption) { eventConsumer = new EventConsumer(node); tailingConsumers.add(eventConsumer); } else { eventConsumer = new EventConsumer(trackingToken); } return eventConsumer; } protected  EmbeddedEventStore(Builder builder); static Builder builder(); @PreDestroy void shutDown(); @Override TrackingEventStream openStream(TrackingToken trackingToken);  }