@Test @Timeout(value = FETCH_DELAY / 10, unit = MILLISECONDS) void testEventPublishedAfterOpeningStreamIsPassedToReaderImmediately() throws Exception { TrackingEventStream stream = testSubject.openStream(null); assertFalse(stream.hasNextAvailable()); DomainEventMessage<?> expected = createEvent(); Thread t = new Thread(() -> { try { assertEquals(expected.getIdentifier(), stream.nextAvailable().getIdentifier()); } catch (InterruptedException e) { fail(); } }); t.start(); testSubject.publish(expected); t.join(); }
@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);  }