// Path: src/main/java/com/oasisdigital/nges/cluster/EventUpdate.java
// public class EventUpdate implements Serializable {
//     private static final long serialVersionUID = 4220177459255722476L;
// 
//     private long eventId;
// 
//     public EventUpdate(long eventId) {
//         this.eventId = eventId;
//     }
// 
//     public long getEventId() {
//         return eventId;
//     }
// }
// 
// Path: src/main/java/com/oasisdigital/nges/cluster/MessageGroup.java
// public interface MessageGroup {
// 
//     /**
//      * Publish a message to the group.
//      */
//     public abstract void publish(Object message);
// 
//     /**
//      * Register a subscriber for messages from the group. See documentation for a particular implementation to
//      * learn more about its specifics.
//      */
//     public abstract void registerSubscriber(Object subscriber);
// 
//     /**
//      * Unregister a subscriber from the group.
//      */
//     public abstract void unregisterSubscriber(Object subscriber);
// 
// }
// 
// Path: src/main/java/com/oasisdigital/nges/event/internal/EventStoreStatusPublisher.java
// public class EventStoreStatusPublisher {
//     private final MessageGroup messageGroup;
// 
//     private long lastEventId = 0;
// 
//     public EventStoreStatusPublisher(MessageGroup messageGroup) {
//         this.messageGroup = messageGroup;
//     }
// 
//     public synchronized void publishLastEventIfChanged(long eventId) {
//         // It's OK to repeat the message (once could be lost, or a listener might join late), but the clock
//         // cannot go backwards.
//         if (lastEventId <= eventId) {
//             messageGroup.publish(new EventUpdate(eventId));
//             lastEventId = eventId;
//         }
//     }
// }
// 
// Path: src/main/java/com/oasisdigital/nges/event/util/LogThrottle.java
// public class LogThrottle {
//     private final Cache<String, Object> cache;
// 
//     public LogThrottle(long duration, TimeUnit timeUnit) {
//         this.cache = CacheBuilder.newBuilder().expireAfterWrite(duration, timeUnit).build();
//     }
// 
//     /**
//      *
//      * @return true if the exception should be throttled (has been written recently).
//      */
//     public boolean throttle(Exception exception) {
//         String exceptionStr = exception.toString();
//         boolean throttled = cache.getIfPresent(exceptionStr) != null;
//         if (!throttled) {
//             cache.put(exceptionStr, new Object());
//         }
//         return throttled;
//     }
// }


import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
import com.oasisdigital.nges.cluster.EventUpdate;
import com.oasisdigital.nges.cluster.MessageGroup;
import com.oasisdigital.nges.event.internal.EventStoreStatusPublisher;
import com.oasisdigital.nges.event.util.LogThrottle;

        this.pollingIntervalMs = pollingIntervalMs;
    }

    synchronized public void initialize() {
        executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(catchAllAndLogWithThrottle(this::publishLastEvent), initialDelayMs,
                pollingIntervalMs, TimeUnit.MILLISECONDS);
    }

    private Runnable catchAllAndLogWithThrottle(Runnable r) {
        LogThrottle throttle = new LogThrottle(1, TimeUnit.HOURS);
        return () -> {
            try {
                r.run();
            } catch (RuntimeException e) {
                if (throttle.throttle(e)) {
                    log.error("Execution failed: " + e);
                } else {
                    log.error("Execution failed", e);
                }
            }
        };
    }

    public void destroy() {
        executor.shutdownNow();
    }

    private void publishLastEvent() {
// Path: src/main/java/com/oasisdigital/nges/cluster/EventUpdate.java
// public class EventUpdate implements Serializable {
//     private static final long serialVersionUID = 4220177459255722476L;
// 
//     private long eventId;
// 
//     public EventUpdate(long eventId) {
//         this.eventId = eventId;
//     }
// 
//     public long getEventId() {
//         return eventId;
//     }
// }
// 
// Path: src/main/java/com/oasisdigital/nges/cluster/MessageGroup.java
// public interface MessageGroup {
// 
//     /**
//      * Publish a message to the group.
//      */
//     public abstract void publish(Object message);
// 
//     /**
//      * Register a subscriber for messages from the group. See documentation for a particular implementation to
//      * learn more about its specifics.
//      */
//     public abstract void registerSubscriber(Object subscriber);
// 
//     /**
//      * Unregister a subscriber from the group.
//      */
//     public abstract void unregisterSubscriber(Object subscriber);
// 
// }
// 
// Path: src/main/java/com/oasisdigital/nges/event/internal/EventStoreStatusPublisher.java
// public class EventStoreStatusPublisher {
//     private final MessageGroup messageGroup;
// 
//     private long lastEventId = 0;
// 
//     public EventStoreStatusPublisher(MessageGroup messageGroup) {
//         this.messageGroup = messageGroup;
//     }
// 
//     public synchronized void publishLastEventIfChanged(long eventId) {
//         // It's OK to repeat the message (once could be lost, or a listener might join late), but the clock
//         // cannot go backwards.
//         if (lastEventId <= eventId) {
//             messageGroup.publish(new EventUpdate(eventId));
//             lastEventId = eventId;
//         }
//     }
// }
// 
// Path: src/main/java/com/oasisdigital/nges/event/util/LogThrottle.java
// public class LogThrottle {
//     private final Cache<String, Object> cache;
// 
//     public LogThrottle(long duration, TimeUnit timeUnit) {
//         this.cache = CacheBuilder.newBuilder().expireAfterWrite(duration, timeUnit).build();
//     }
// 
//     /**
//      *
//      * @return true if the exception should be throttled (has been written recently).
//      */
//     public boolean throttle(Exception exception) {
//         String exceptionStr = exception.toString();
//         boolean throttled = cache.getIfPresent(exceptionStr) != null;
//         if (!throttled) {
//             cache.put(exceptionStr, new Object());
//         }
//         return throttled;
//     }
// }


// Path: src/main/java/com/oasisdigital/nges/event/EventStoreStatusHeartbeat.java
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
import com.oasisdigital.nges.cluster.EventUpdate;
import com.oasisdigital.nges.cluster.MessageGroup;
import com.oasisdigital.nges.event.internal.EventStoreStatusPublisher;
import com.oasisdigital.nges.event.util.LogThrottle;


        this.pollingIntervalMs = pollingIntervalMs;
    }

    synchronized public void initialize() {
        executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(catchAllAndLogWithThrottle(this::publishLastEvent), initialDelayMs,
                pollingIntervalMs, TimeUnit.MILLISECONDS);
    }

    private Runnable catchAllAndLogWithThrottle(Runnable r) {
        LogThrottle throttle = new LogThrottle(1, TimeUnit.HOURS);
        return () -> {
            try {
                r.run();
            } catch (RuntimeException e) {
                if (throttle.throttle(e)) {
                    log.error("Execution failed: " + e);
                } else {
                    log.error("Execution failed", e);
                }
            }
        };
    }

    public void destroy() {
        executor.shutdownNow();
    }

    private void publishLastEvent() {
        eventStore.getLastEventId().ifPresent(publisher::publishLastEventIfChanged);