@Test @Timeout(value = 5) void testEventIsFetchedFromCacheWhenFetchedASecondTime() throws Exception { CountDownLatch lock = new CountDownLatch(2); List<TrackedEventMessage<?>> events = new CopyOnWriteArrayList<>(); Thread t = new Thread(() -> testSubject.openStream(null).asStream().limit(2).forEach(event -> { lock.countDown(); events.add(event); })); t.start(); assertFalse(lock.await(100, MILLISECONDS)); testSubject.publish(createEvents(2)); t.join(); TrackedEventMessage<?> second = testSubject.openStream(events.get(0).trackingToken()).nextAvailable(); assertSame(events.get(1), second); }
@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);  }