/*
 * Copyright 2015 Open Networking Laboratory
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.onosproject.store.ecmap;

import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.commons.lang3.tuple.Pair;
import org.onlab.util.KryoNamespace;
import org.onlab.util.SlidingWindowCounter;
import org.onosproject.cluster.ClusterService;
import org.onosproject.cluster.ControllerNode;
import org.onosproject.cluster.NodeId;
import org.onosproject.store.Timestamp;
import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
import org.onosproject.store.cluster.messaging.ClusterMessage;
import org.onosproject.store.cluster.messaging.ClusterMessageHandler;
import org.onosproject.store.cluster.messaging.MessageSubject;
import org.onosproject.store.impl.ClockService;
import org.onosproject.store.impl.Timestamped;
import org.onosproject.store.impl.WallClockTimestamp;
import org.onosproject.store.serializers.KryoSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
import static org.onlab.util.BoundedThreadPool.newFixedThreadPool;
import static org.onlab.util.Tools.groupedThreads;

/**
 * Distributed Map implementation which uses optimistic replication and gossip
 * based techniques to provide an eventually consistent data store.
 */
public class EventuallyConsistentMapImpl<K, V>
        implements EventuallyConsistentMap<K, V> {

    private static final Logger log = LoggerFactory.getLogger(EventuallyConsistentMapImpl.class);

    private final ConcurrentMap<K, Timestamped<V>> items;
    private final ConcurrentMap<K, Timestamp> removedItems;

    private final ClusterService clusterService;
    private final ClusterCommunicationService clusterCommunicator;
    private final KryoSerializer serializer;

    private final ClockService<K, V> clockService;

    private final MessageSubject updateMessageSubject;
    private final MessageSubject removeMessageSubject;
    private final MessageSubject antiEntropyAdvertisementSubject;

    private final Set<EventuallyConsistentMapListener<K, V>> listeners
            = new CopyOnWriteArraySet<>();

    private final ExecutorService executor;

    private final ScheduledExecutorService backgroundExecutor;
    private final BiFunction<K, V, Iterable<NodeId>> peerUpdateFunction;

    private ExecutorService broadcastMessageExecutor;

    private volatile boolean destroyed = false;
    private static final String ERROR_DESTROYED = " map is already destroyed";
    private final String destroyedMessage;

    private static final String ERROR_NULL_KEY = "Key cannot be null";
    private static final String ERROR_NULL_VALUE = "Null values are not allowed";

    // TODO: Make these anti-entropy params configurable
    private long initialDelaySec = 5;
    private long periodSec = 5;
    private boolean lightweightAntiEntropy = true;

    private static final int WINDOW_SIZE = 5;
    private static final int HIGH_LOAD_THRESHOLD = 0;
    private static final int LOAD_WINDOW = 2;
    SlidingWindowCounter counter = new SlidingWindowCounter(WINDOW_SIZE);
    AtomicLong operations = new AtomicLong();

    /**
     * Creates a new eventually consistent map shared amongst multiple instances.
     * <p>
     * Each map is identified by a string map name. EventuallyConsistentMapImpl
     * objects in different JVMs that use the same map name will form a
     * distributed map across JVMs (provided the cluster service is aware of
     * both nodes).
     * </p>
     * <p>
     * The client is expected to provide an
     * {@link org.onlab.util.KryoNamespace.Builder} with which all classes that
     * will be stored in this map have been registered (including referenced
     * classes). This serializer will be used to serialize both K and V for
     * inter-node notifications.
     * </p>
     * <p>
     * The client must provide an {@link org.onosproject.store.impl.ClockService}
     * which can generate timestamps for a given key. The clock service is free
     * to generate timestamps however it wishes, however these timestamps will
     * be used to serialize updates to the map so they must be strict enough
     * to ensure updates are properly ordered for the use case (i.e. in some
     * cases wallclock time will suffice, whereas in other cases logical time
     * will be necessary).
     * </p>
     *
     * @param mapName             a String identifier for the map.
     * @param clusterService      the cluster service
     * @param clusterCommunicator the cluster communications service
     * @param serializerBuilder   a Kryo namespace builder that can serialize
     *                            both K and V
     * @param clockService        a clock service able to generate timestamps
     *                            for K
     * @param peerUpdateFunction  function that provides a set of nodes to immediately
     *                            update to when there writes to the map
     */
    public EventuallyConsistentMapImpl(String mapName,
                                       ClusterService clusterService,
                                       ClusterCommunicationService clusterCommunicator,
                                       KryoNamespace.Builder serializerBuilder,
                                       ClockService<K, V> clockService,
                                       BiFunction<K, V, Iterable<NodeId>> peerUpdateFunction) {
        this.clusterService = checkNotNull(clusterService);
        this.clusterCommunicator = checkNotNull(clusterCommunicator);
        this.peerUpdateFunction = checkNotNull(peerUpdateFunction);

        serializer = createSerializer(checkNotNull(serializerBuilder));
        destroyedMessage = mapName + ERROR_DESTROYED;

        this.clockService = checkNotNull(clockService);

        items = new ConcurrentHashMap<>();
        removedItems = new ConcurrentHashMap<>();

        // should be a normal executor; it's used for receiving messages
        //TODO make # of threads configurable
        executor = Executors.newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-fg-%d"));

        // sending executor; should be capped
        //TODO make # of threads configurable
        broadcastMessageExecutor = //newSingleThreadExecutor(groupedThreads("onos/ecm", mapName + "-notify"));
                newFixedThreadPool(4, groupedThreads("onos/ecm", mapName + "-notify"));

        backgroundExecutor =
                //FIXME anti-entropy can take >60 seconds and it blocks fg workers
                // ... dropping minPriority to try to help until this can be parallel
                newSingleThreadScheduledExecutor(//minPriority(
                                                 groupedThreads("onos/ecm", mapName + "-bg-%d"))/*)*/;

        // start anti-entropy thread
        //TODO disable anti-entropy for now in testing (it is unstable)
        backgroundExecutor.scheduleAtFixedRate(new SendAdvertisementTask(),
                                               initialDelaySec, periodSec,
                                               TimeUnit.SECONDS);

        updateMessageSubject = new MessageSubject("ecm-" + mapName + "-update");
        clusterCommunicator.addSubscriber(updateMessageSubject,
                                          new InternalPutEventListener(), executor);
        removeMessageSubject = new MessageSubject("ecm-" + mapName + "-remove");
        clusterCommunicator.addSubscriber(removeMessageSubject,
                                          new InternalRemoveEventListener(), executor);
        antiEntropyAdvertisementSubject = new MessageSubject("ecm-" + mapName + "-anti-entropy");
        clusterCommunicator.addSubscriber(antiEntropyAdvertisementSubject,
                                          new InternalAntiEntropyListener(), backgroundExecutor);
    }

    /**
     * Creates a new eventually consistent map shared amongst multiple instances.
     * <p>
     * Take a look at the other constructor for usage information. The only difference
     * is that a BiFunction is provided that returns all nodes in the cluster, so
     * all nodes will be sent write updates immediately.
     * </p>
     *
     * @param mapName             a String identifier for the map.
     * @param clusterService      the cluster service
     * @param clusterCommunicator the cluster communications service
     * @param serializerBuilder   a Kryo namespace builder that can serialize
     *                            both K and V
     * @param clockService        a clock service able to generate timestamps
     *                            for K
     */
    public EventuallyConsistentMapImpl(String mapName,
                                       ClusterService clusterService,
                                       ClusterCommunicationService clusterCommunicator,
                                       KryoNamespace.Builder serializerBuilder,
                                       ClockService<K, V> clockService) {
        this(mapName, clusterService, clusterCommunicator, serializerBuilder, clockService,
             (key, value) -> clusterService.getNodes().stream()
                     .map(ControllerNode::id)
                     .filter(nodeId -> !nodeId.equals(clusterService.getLocalNode().id()))
                     .collect(Collectors.toList()));
    }

    private KryoSerializer createSerializer(KryoNamespace.Builder builder) {
        return new KryoSerializer() {
            @Override
            protected void setupKryoPool() {
                // Add the map's internal helper classes to the user-supplied serializer
                serializerPool = builder
                        .register(WallClockTimestamp.class)
                        .register(PutEntry.class)
                        .register(RemoveEntry.class)
                        .register(ArrayList.class)
                        .register(InternalPutEvent.class)
                        .register(InternalRemoveEvent.class)
                        .register(AntiEntropyAdvertisement.class)
                        .register(HashMap.class)
                        .build();
            }
        };
    }

    /**
     * Sets the executor to use for broadcasting messages and returns this
     * instance for method chaining.
     *
     * @param executor executor service
     * @return this instance
     */
    public EventuallyConsistentMapImpl<K, V> withBroadcastMessageExecutor(ExecutorService executor) {
        checkNotNull(executor, "Null executor");
        broadcastMessageExecutor = executor;
        return this;
    }

    @Override
    public int size() {
        checkState(!destroyed, destroyedMessage);
        return items.size();
    }

    @Override
    public boolean isEmpty() {
        checkState(!destroyed, destroyedMessage);
        return items.isEmpty();
    }

    @Override
    public boolean containsKey(K key) {
        checkState(!destroyed, destroyedMessage);
        checkNotNull(key, ERROR_NULL_KEY);
        return items.containsKey(key);
    }

    @Override
    public boolean containsValue(V value) {
        checkState(!destroyed, destroyedMessage);
        checkNotNull(value, ERROR_NULL_VALUE);

        return items.values().stream()
                .anyMatch(timestamped -> timestamped.value().equals(value));
    }

    @Override
    public V get(K key) {
        checkState(!destroyed, destroyedMessage);
        checkNotNull(key, ERROR_NULL_KEY);

        Timestamped<V> value = items.get(key);
        if (value != null) {
            return value.value();
        }
        return null;
    }

    @Override
    public void put(K key, V value) {
        checkState(!destroyed, destroyedMessage);
        checkNotNull(key, ERROR_NULL_KEY);
        checkNotNull(value, ERROR_NULL_VALUE);

        Timestamp timestamp = clockService.getTimestamp(key, value);

        if (putInternal(key, value, timestamp)) {
            notifyPeers(new InternalPutEvent<>(key, value, timestamp),
                        peerUpdateFunction.apply(key, value));
            notifyListeners(new EventuallyConsistentMapEvent<>(
                    EventuallyConsistentMapEvent.Type.PUT, key, value));
        }
    }

    private boolean putInternal(K key, V value, Timestamp timestamp) {
        counter.incrementCount();
        Timestamp removed = removedItems.get(key);
        if (removed != null && removed.isNewerThan(timestamp)) {
            log.debug("ecmap - removed was newer {}", value);
            return false;
        }

        final MutableBoolean updated = new MutableBoolean(false);

        items.compute(key, (k, existing) -> {
            if (existing != null && existing.isNewerThan(timestamp)) {
                updated.setFalse();
                return existing;
            } else {
                updated.setTrue();
                return new Timestamped<>(value, timestamp);
            }
            });

        boolean success = updated.booleanValue();
        if (!success) {
            log.debug("ecmap - existing was newer {}", value);
        }

        if (success && removed != null) {
            removedItems.remove(key, removed);
        }
        return success;
    }

    @Override
    public void remove(K key) {
        checkState(!destroyed, destroyedMessage);
        checkNotNull(key, ERROR_NULL_KEY);

        // TODO prevent calls here if value is important for timestamp
        Timestamp timestamp = clockService.getTimestamp(key, null);

        if (removeInternal(key, timestamp)) {
            notifyPeers(new InternalRemoveEvent<>(key, timestamp),
                        peerUpdateFunction.apply(key, null));
            notifyListeners(new EventuallyConsistentMapEvent<>(
                    EventuallyConsistentMapEvent.Type.REMOVE, key, null));
        }
    }

    private boolean removeInternal(K key, Timestamp timestamp) {
        counter.incrementCount();
        final MutableBoolean updated = new MutableBoolean(false);

        items.compute(key, (k, existing) -> {
            if (existing != null && existing.isNewerThan(timestamp)) {
                updated.setFalse();
                return existing;
            } else {
                updated.setTrue();
                // remove from items map
                return null;
            }
            });

        if (updated.isFalse()) {
            return false;
        }

        Timestamp removedTimestamp = removedItems.get(key);
        if (removedTimestamp == null) {
            return removedItems.putIfAbsent(key, timestamp) == null;
        } else if (timestamp.isNewerThan(removedTimestamp)) {
            return removedItems.replace(key, removedTimestamp, timestamp);
        } else {
            return false;
        }
    }

    @Override
    public void remove(K key, V value) {
        checkState(!destroyed, destroyedMessage);
        checkNotNull(key, ERROR_NULL_KEY);
        checkNotNull(value, ERROR_NULL_VALUE);

        Timestamp timestamp = clockService.getTimestamp(key, value);

        if (removeInternal(key, timestamp)) {
            notifyPeers(new InternalRemoveEvent<>(key, timestamp),
                        peerUpdateFunction.apply(key, value));
            notifyListeners(new EventuallyConsistentMapEvent<>(
                    EventuallyConsistentMapEvent.Type.REMOVE, key, value));
        }
    }

    @Override
    public void putAll(Map<? extends K, ? extends V> m) {
        checkState(!destroyed, destroyedMessage);

        List<PutEntry<K, V>> updates = new ArrayList<>(m.size());

        for (Map.Entry<? extends K, ? extends V> entry : m.entrySet()) {
            K key = entry.getKey();
            V value = entry.getValue();

            checkNotNull(key, ERROR_NULL_KEY);
            checkNotNull(value, ERROR_NULL_VALUE);

            Timestamp timestamp = clockService.getTimestamp(key, value);

            if (putInternal(key, value, timestamp)) {
                updates.add(new PutEntry<>(key, value, timestamp));
            }
        }

        if (!updates.isEmpty()) {
            broadcastMessage(updateMessageSubject, new InternalPutEvent<>(updates));

            for (PutEntry<K, V> entry : updates) {
                EventuallyConsistentMapEvent<K, V> externalEvent =
                        new EventuallyConsistentMapEvent<>(
                                EventuallyConsistentMapEvent.Type.PUT, entry.key(),
                                entry.value());
                notifyListeners(externalEvent);
            }
        }
    }

    @Override
    public void clear() {
        checkState(!destroyed, destroyedMessage);

        List<RemoveEntry<K>> removed = new ArrayList<>(items.size());

        for (K key : items.keySet()) {
            // TODO also this is not applicable if value is important for timestamp?
            Timestamp timestamp = clockService.getTimestamp(key, null);

            if (removeInternal(key, timestamp)) {
                removed.add(new RemoveEntry<>(key, timestamp));
            }
        }

        if (!removed.isEmpty()) {
            broadcastMessage(removeMessageSubject, new InternalRemoveEvent<>(removed));

            for (RemoveEntry<K> entry : removed) {
                EventuallyConsistentMapEvent<K, V> externalEvent
                        = new EventuallyConsistentMapEvent<>(
                        EventuallyConsistentMapEvent.Type.REMOVE, entry.key(),
                        null);
                notifyListeners(externalEvent);
            }
        }
    }

    @Override
    public Set<K> keySet() {
        checkState(!destroyed, destroyedMessage);

        return items.keySet();
    }

    @Override
    public Collection<V> values() {
        checkState(!destroyed, destroyedMessage);

        return items.values().stream()
                .map(Timestamped::value)
                .collect(Collectors.toList());
    }

    @Override
    public Set<Map.Entry<K, V>> entrySet() {
        checkState(!destroyed, destroyedMessage);

        return items.entrySet().stream()
                .map(e -> Pair.of(e.getKey(), e.getValue().value()))
                .collect(Collectors.toSet());
    }

    @Override
    public void addListener(EventuallyConsistentMapListener<K, V> listener) {
        checkState(!destroyed, destroyedMessage);

        listeners.add(checkNotNull(listener));
    }

    @Override
    public void removeListener(EventuallyConsistentMapListener<K, V> listener) {
        checkState(!destroyed, destroyedMessage);

        listeners.remove(checkNotNull(listener));
    }

    @Override
    public void destroy() {
        destroyed = true;

        executor.shutdown();
        backgroundExecutor.shutdown();
        broadcastMessageExecutor.shutdown();

        listeners.clear();

        clusterCommunicator.removeSubscriber(updateMessageSubject);
        clusterCommunicator.removeSubscriber(removeMessageSubject);
        clusterCommunicator.removeSubscriber(antiEntropyAdvertisementSubject);
    }

    private void notifyListeners(EventuallyConsistentMapEvent<K, V> event) {
        for (EventuallyConsistentMapListener<K, V> listener : listeners) {
            listener.event(event);
        }
    }

    private void notifyPeers(InternalPutEvent event, Iterable<NodeId> peers) {
        // FIXME extremely memory expensive when we are overrun
//        broadcastMessageExecutor.execute(() -> broadcastMessage(updateMessageSubject, event));
        multicastMessage(updateMessageSubject, event, peers);
    }

    private void notifyPeers(InternalRemoveEvent event, Iterable<NodeId> peers) {
        // FIXME extremely memory expensive when we are overrun
//        broadcastMessageExecutor.execute(() -> broadcastMessage(removeMessageSubject, event));
        multicastMessage(removeMessageSubject, event, peers);
    }

    private void multicastMessage(MessageSubject subject, Object event, Iterable<NodeId> peers) {
        // FIXME can we parallelize the serialization... use the caller???
        ClusterMessage message = new ClusterMessage(
                clusterService.getLocalNode().id(),
                subject,
                serializer.encode(event));
        broadcastMessageExecutor.execute(() -> clusterCommunicator.multicast(message, peers));
//        clusterCommunicator.broadcast(message);
    }

    private void broadcastMessage(MessageSubject subject, Object event) {
        // FIXME can we parallelize the serialization... use the caller???
        ClusterMessage message = new ClusterMessage(
                clusterService.getLocalNode().id(),
                subject,
                serializer.encode(event));
        broadcastMessageExecutor.execute(() -> clusterCommunicator.broadcast(message));
//        clusterCommunicator.broadcast(message);
    }

    private void unicastMessage(NodeId peer, MessageSubject subject, Object event) {
        ClusterMessage message = new ClusterMessage(
                clusterService.getLocalNode().id(),
                subject,
                serializer.encode(event));
//        clusterCommunicator.unicast(message, peer);
        broadcastMessageExecutor.execute(() -> clusterCommunicator.unicast(message, peer));
    }

    private boolean underHighLoad() {
        return counter.get(LOAD_WINDOW) > HIGH_LOAD_THRESHOLD;
    }

    private final class SendAdvertisementTask implements Runnable {
        @Override
        public void run() {
            if (Thread.currentThread().isInterrupted()) {
                log.info("Interrupted, quitting");
                return;
            }

            if (underHighLoad()) {
                return;
            }

            try {
                final NodeId self = clusterService.getLocalNode().id();
                Set<ControllerNode> nodes = clusterService.getNodes();

                List<NodeId> nodeIds = nodes.stream()
                        .map(ControllerNode::id)
                        .collect(Collectors.toList());

                if (nodeIds.size() == 1 && nodeIds.get(0).equals(self)) {
                    log.trace("No other peers in the cluster.");
                    return;
                }

                NodeId peer;
                do {
                    int idx = RandomUtils.nextInt(0, nodeIds.size());
                    peer = nodeIds.get(idx);
                } while (peer.equals(self));

                if (Thread.currentThread().isInterrupted()) {
                    log.info("Interrupted, quitting");
                    return;
                }

                AntiEntropyAdvertisement<K> ad = createAdvertisement();

                // TODO check the return value?
                unicastMessage(peer, antiEntropyAdvertisementSubject, ad);
                // error log: log.debug("Failed to send anti-entropy advertisement to {}", peer);
            } catch (Exception e) {
                // Catch all exceptions to avoid scheduled task being suppressed.
                log.error("Exception thrown while sending advertisement", e);
            }
        }
    }

    private AntiEntropyAdvertisement<K> createAdvertisement() {
        final NodeId self = clusterService.getLocalNode().id();

        Map<K, Timestamp> timestamps = new HashMap<>(items.size());

        items.forEach((key, value) -> timestamps.put(key, value.timestamp()));

        Map<K, Timestamp> tombstones = new HashMap<>(removedItems);

        return new AntiEntropyAdvertisement<>(self, timestamps, tombstones);
    }

    private void handleAntiEntropyAdvertisement(AntiEntropyAdvertisement<K> ad) {
        List<EventuallyConsistentMapEvent<K, V>> externalEvents;

        externalEvents = antiEntropyCheckLocalItems(ad);

        antiEntropyCheckLocalRemoved(ad);

        if (!lightweightAntiEntropy) {
            externalEvents.addAll(antiEntropyCheckRemoteRemoved(ad));

            // if remote ad has something unknown, actively sync
            for (K key : ad.timestamps().keySet()) {
                if (!items.containsKey(key)) {
                    // Send the advertisement back if this peer is out-of-sync
                    final NodeId sender = ad.sender();
                    AntiEntropyAdvertisement<K> myAd = createAdvertisement();
                    // TODO check the return value?
                    unicastMessage(sender, antiEntropyAdvertisementSubject, myAd);
                    // error log: log.debug("Failed to send reactive anti-entropy advertisement to {}", sender);
                    break;
                }
            }
        }
        externalEvents.forEach(this::notifyListeners);
    }

    /**
     * Checks if any of the remote's live items or tombstones are out of date
     * according to our local live item list, or if our live items are out of
     * date according to the remote's tombstone list.
     * If the local copy is more recent, it will be pushed to the remote. If the
     * remote has a more recent remove, we apply that to the local state.
     *
     * @param ad remote anti-entropy advertisement
     * @return list of external events relating to local operations performed
     */
    private List<EventuallyConsistentMapEvent<K, V>> antiEntropyCheckLocalItems(
            AntiEntropyAdvertisement<K> ad) {
        final List<EventuallyConsistentMapEvent<K, V>> externalEvents
                = new LinkedList<>();
        final NodeId sender = ad.sender();

        final List<PutEntry<K, V>> updatesToSend = new ArrayList<>();

        for (Map.Entry<K, Timestamped<V>> item : items.entrySet()) {
            K key = item.getKey();
            Timestamped<V> localValue = item.getValue();

            Timestamp remoteTimestamp = ad.timestamps().get(key);
            if (remoteTimestamp == null) {
                remoteTimestamp = ad.tombstones().get(key);
            }
            if (remoteTimestamp == null || localValue
                    .isNewerThan(remoteTimestamp)) {
                // local value is more recent, push to sender
                updatesToSend
                        .add(new PutEntry<>(key, localValue.value(),
                                            localValue.timestamp()));
            }

            Timestamp remoteDeadTimestamp = ad.tombstones().get(key);
            if (remoteDeadTimestamp != null &&
                    remoteDeadTimestamp.isNewerThan(localValue.timestamp())) {
                // sender has a more recent remove
                if (removeInternal(key, remoteDeadTimestamp)) {
                    externalEvents.add(new EventuallyConsistentMapEvent<>(
                            EventuallyConsistentMapEvent.Type.REMOVE, key, null));
                }
            }
        }

        // Send all updates to the peer at once
        if (!updatesToSend.isEmpty()) {
            // TODO check the return value?
            unicastMessage(sender, updateMessageSubject,
                           new InternalPutEvent<>(updatesToSend));
            //error log: log.warn("Failed to send advertisement response", e);
        }

        return externalEvents;
    }

    /**
     * Checks if any items in the remote live list are out of date according
     * to our tombstone list. If we find we have a more up to date tombstone,
     * we'll send it to the remote.
     *
     * @param ad remote anti-entropy advertisement
     */
    private void antiEntropyCheckLocalRemoved(AntiEntropyAdvertisement<K> ad) {
        final NodeId sender = ad.sender();

        final List<RemoveEntry<K>> removesToSend = new ArrayList<>();

        for (Map.Entry<K, Timestamp> dead : removedItems.entrySet()) {
            K key = dead.getKey();
            Timestamp localDeadTimestamp = dead.getValue();

            Timestamp remoteLiveTimestamp = ad.timestamps().get(key);
            if (remoteLiveTimestamp != null
                    && localDeadTimestamp.isNewerThan(remoteLiveTimestamp)) {
                // sender has zombie, push remove
                removesToSend
                        .add(new RemoveEntry<>(key, localDeadTimestamp));
            }
        }

        // Send all removes to the peer at once
        if (!removesToSend.isEmpty()) {
            // TODO check the return value
            unicastMessage(sender, removeMessageSubject,
                           new InternalRemoveEvent<>(removesToSend));
            // error log: log.warn("Failed to send advertisement response", e);
        }
    }

    /**
     * Checks if any of the local live items are out of date according to the
     * remote's tombstone advertisements. If we find a local item is out of date,
     * we'll apply the remove operation to the local state.
     *
     * @param ad remote anti-entropy advertisement
     * @return list of external events relating to local operations performed
     */
    private List<EventuallyConsistentMapEvent<K, V>>
    antiEntropyCheckRemoteRemoved(AntiEntropyAdvertisement<K> ad) {
        final List<EventuallyConsistentMapEvent<K, V>> externalEvents
                = new LinkedList<>();

        for (Map.Entry<K, Timestamp> remoteDead : ad.tombstones().entrySet()) {
            K key = remoteDead.getKey();
            Timestamp remoteDeadTimestamp = remoteDead.getValue();

            Timestamped<V> local = items.get(key);
            Timestamp localDead = removedItems.get(key);
            if (local != null && remoteDeadTimestamp.isNewerThan(
                    local.timestamp())) {
                // If the remote has a more recent tombstone than either our local
                // value, then do a remove with their timestamp
                if (removeInternal(key, remoteDeadTimestamp)) {
                    externalEvents.add(new EventuallyConsistentMapEvent<>(
                            EventuallyConsistentMapEvent.Type.REMOVE, key, null));
                }
            } else if (localDead != null && remoteDeadTimestamp.isNewerThan(
                    localDead)) {
                // If the remote has a more recent tombstone than us, update ours
                // to their timestamp
                removeInternal(key, remoteDeadTimestamp);
            }
        }

        return externalEvents;
    }

    private final class InternalAntiEntropyListener
            implements ClusterMessageHandler {

        @Override
        public void handle(ClusterMessage message) {
            log.trace("Received anti-entropy advertisement from peer: {}",
                      message.sender());
            AntiEntropyAdvertisement<K> advertisement = serializer.decode(message.payload());
            try {
                if (!underHighLoad()) {
                    handleAntiEntropyAdvertisement(advertisement);
                }
            } catch (Exception e) {
                log.warn("Exception thrown handling advertisements", e);
            }
        }
    }

    private final class InternalPutEventListener implements
            ClusterMessageHandler {
        @Override
        public void handle(ClusterMessage message) {
            log.debug("Received put event from peer: {}", message.sender());
            InternalPutEvent<K, V> event = serializer.decode(message.payload());

            try {
                for (PutEntry<K, V> entry : event.entries()) {
                    K key = entry.key();
                    V value = entry.value();
                    Timestamp timestamp = entry.timestamp();

                    if (putInternal(key, value, timestamp)) {
                        EventuallyConsistentMapEvent<K, V> externalEvent =
                                new EventuallyConsistentMapEvent<>(
                                        EventuallyConsistentMapEvent.Type.PUT, key,
                                        value);
                        notifyListeners(externalEvent);
                    }
                }
            } catch (Exception e) {
                log.warn("Exception thrown handling put", e);
            }
        }
    }

    private final class InternalRemoveEventListener implements
            ClusterMessageHandler {
        @Override
        public void handle(ClusterMessage message) {
            log.debug("Received remove event from peer: {}", message.sender());
            InternalRemoveEvent<K> event = serializer.decode(message.payload());
            try {
                for (RemoveEntry<K> entry : event.entries()) {
                    K key = entry.key();
                    Timestamp timestamp = entry.timestamp();

                    if (removeInternal(key, timestamp)) {
                        EventuallyConsistentMapEvent<K, V> externalEvent
                        = new EventuallyConsistentMapEvent<>(
                                EventuallyConsistentMapEvent.Type.REMOVE,
                                key, null);
                        notifyListeners(externalEvent);
                    }
                }
            } catch (Exception e) {
                log.warn("Exception thrown handling remove", e);
            }
        }
    }

}