@Test void testEventStreamKeepsReturningEventsIfEventConsumptionOptimizationIsSwitchedOff() throws InterruptedException { boolean doNotOptimizeEventConsumption = false; newTestSubject(CACHED_EVENTS, FETCH_DELAY, CLEANUP_DELAY, doNotOptimizeEventConsumption); TrackingEventStream eventStream = testSubject.openStream(null); assertFalse(eventStream.hasNextAvailable()); testSubject.publish(createEvents(5)); assertTrue(eventStream.hasNextAvailable()); while (eventStream.hasNextAvailable()) { eventStream.nextAvailable(); } assertFalse(eventStream.hasNextAvailable()); }
@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);  }