ReplayingCluster replayingCluster = new ReplayingCluster(new SimpleCluster("simple"), eventStore,
new NoTransactionManager(), 0,
new BackloggingIncomingMessageHandler());
// we initialize an event bus that contains our replaying cluster
EventBus eventBus = new ClusteringEventBus(new DefaultClusterSelector(replayingCluster));
// we subscribe our two listeners to the Event Bus
AnnotationEventListenerAdapter.subscribe(new ThreadPrintingEventListener(), eventBus);
AnnotationEventListenerAdapter.subscribe(new AnotherThreadPrintingEventListener(), eventBus);
// we append some events to simulate a full event store
final DomainEventMessage[] domainEventMessages = {
new GenericDomainEventMessage<ToDoItemCreatedEvent>(
"todo1", 0, new ToDoItemCreatedEvent("todo1", "Need to do something")),
new GenericDomainEventMessage<ToDoItemCreatedEvent>(
"todo2", 0, new ToDoItemCreatedEvent("todo2", "Another thing to do")),
new GenericDomainEventMessage<ToDoItemCompletedEvent>("todo2", 0, new ToDoItemCompletedEvent("todo2"))
};
eventStore.appendEvents("mock", new SimpleDomainEventStream(domainEventMessages));
// we create an executor service with a single thread and start the replay as an asynchronous process
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<Void> future = replayingCluster.startReplay(executor);
// we want to wait for the cluster to have switched to replay mode, so we can send some messages to it.
// if we were to publish events right away, there is a big chance the Cluster didn't switch to replay mode, yet.
waitForReplayToHaveStarted(replayingCluster);
// this is a new event, so it should be backlogged and handled at the end of the replay
eventBus.publish(asEventMessage(new ToDoItemCreatedEvent("todo3", "Came in just now...")));
// this message is also part of the replay, and should therefore not be handled twice.
eventBus.publish(domainEventMessages[2]);
// we wait (at most 10 seconds) for the replay to complete.
future.get(10, TimeUnit.SECONDS);
// and we publish another event to show that it's handled in the calling thread
eventBus.publish(asEventMessage(new ToDoItemDeadlineExpiredEvent("todo1")));
// we want to shutdown the executor, to get a proper JVM shutdown
executor.shutdown();
}