/*
 * 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 com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.commons.lang3.tuple.Pair;
import org.onlab.util.AbstractAccumulator;
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.Timer;
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 antiEntropyAdvertisementSubject;

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

    private final ExecutorService executor;

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

    private ExecutorService communicationExecutor;
    private Map<NodeId, EventAccumulator> senderPending;

    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 boolean tombstonesDisabled = false;

    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, Collection<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
        //TODO this probably doesn't need to be bounded anymore
        communicationExecutor =
                newFixedThreadPool(8, groupedThreads("onos/ecm", mapName + "-publish-%d"));
        senderPending = Maps.newConcurrentMap();

        backgroundExecutor =
                newSingleThreadScheduledExecutor(groupedThreads("onos/ecm", mapName + "-bg-%d"));

        // start anti-entropy thread
        backgroundExecutor.scheduleAtFixedRate(new SendAdvertisementTask(),
                                               initialDelaySec, periodSec,
                                               TimeUnit.SECONDS);

        updateMessageSubject = new MessageSubject("ecm-" + mapName + "-update");
        clusterCommunicator.addSubscriber(updateMessageSubject,
                                          new InternalEventListener(), 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()));
    }

    public EventuallyConsistentMapImpl<K, V> withTombstonesDisabled(boolean status) {
        tombstonesDisabled = status;
        return this;
    }

    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(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");
        communicationExecutor = 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 PutEntry<>(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 RemoveEntry<>(key, timestamp),
                        peerUpdateFunction.apply(key, null));
            notifyListeners(new EventuallyConsistentMapEvent<>(
                    EventuallyConsistentMapEvent.Type.REMOVE, key, null));
        }
    }

    private boolean removeInternal(K key, Timestamp timestamp) {
        if (timestamp == null) {
            return false;
        }

        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;
        }

        if (!tombstonesDisabled) {
            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;
            }
        }

        return updated.booleanValue();
    }

    @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 RemoveEntry<>(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);
        m.forEach(this::put);
    }

    @Override
    public void clear() {
        checkState(!destroyed, destroyedMessage);
        items.forEach((key, value) -> remove(key));
    }

    @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();
        communicationExecutor.shutdown();

        listeners.clear();

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

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

    private void notifyPeers(PutEntry<K, V> event, Collection<NodeId> peers) {
        queueUpdate(event, peers);
    }

    private void notifyPeers(RemoveEntry<K, V> event, Collection<NodeId> peers) {
        queueUpdate(event, peers);
    }

    private void queueUpdate(AbstractEntry<K, V> event, Collection<NodeId> peers) {
        if (peers == null) {
            // we have no friends :(
            return;
        }
        peers.forEach(node ->
              senderPending.computeIfAbsent(node, unusedKey -> new EventAccumulator(node)).add(event)
        );
    }

    private boolean unicastMessage(NodeId peer, MessageSubject subject, Object event) {
        ClusterMessage message = new ClusterMessage(
                clusterService.getLocalNode().id(),
                subject,
                serializer.encode(event));
        return clusterCommunicator.unicast(message, peer);
        // Note: we had this flipped before...
//        communicationExecutor.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();

                if (!unicastMessage(peer, antiEntropyAdvertisementSubject, ad)) {
                    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();
                    if (!unicastMessage(sender, antiEntropyAdvertisementSubject, myAd)) {
                        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();

        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
                queueUpdate(new PutEntry<>(key, localValue.value(),
                                            localValue.timestamp()), ImmutableList.of(sender));
            }

            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));
                }
            }
        }

        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();

        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
                queueUpdate(new RemoveEntry<>(key, localDeadTimestamp), ImmutableList.of(sender));
            }
        }
    }

    /**
     * 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 InternalEventListener implements
            ClusterMessageHandler {
        @Override
        public void handle(ClusterMessage message) {
            log.debug("Received update event from peer: {}", message.sender());
            Collection<AbstractEntry<K, V>> events = serializer.decode(message.payload());

            try {
                // TODO clean this for loop up
                for (AbstractEntry<K, V> entry : events) {
                    final K key = entry.key();
                    final V value;
                    final Timestamp timestamp = entry.timestamp();
                    final EventuallyConsistentMapEvent.Type type;
                    if (entry instanceof PutEntry) {
                        PutEntry<K, V> putEntry = (PutEntry<K, V>) entry;
                        value = putEntry.value();
                        type = EventuallyConsistentMapEvent.Type.PUT;
                    } else if (entry instanceof RemoveEntry) {
                        type = EventuallyConsistentMapEvent.Type.REMOVE;
                        value = null;
                    } else {
                        throw new IllegalStateException("Unknown entry type " + entry.getClass());
                    }

                    boolean success;
                    switch (type) {
                        case PUT:
                            success = putInternal(key, value, timestamp);
                            break;
                        case REMOVE:
                            success = removeInternal(key, timestamp);
                            break;
                        default:
                            success = false;
                    }
                    if (success) {
                        notifyListeners(new EventuallyConsistentMapEvent<>(type, key, value));
                    }
                }
            } catch (Exception e) {
                log.warn("Exception thrown handling put", e);
            }
        }
    }

    // TODO pull this into the class if this gets pulled out...
    private static final int DEFAULT_MAX_EVENTS = 1000;
    private static final int DEFAULT_MAX_IDLE_MS = 10;
    private static final int DEFAULT_MAX_BATCH_MS = 50;
    private static final Timer TIMER = new Timer("onos-ecm-sender-events");

    private final class EventAccumulator extends AbstractAccumulator<AbstractEntry<K, V>> {

        private final NodeId peer;

        private EventAccumulator(NodeId peer) {
            super(TIMER, DEFAULT_MAX_EVENTS, DEFAULT_MAX_BATCH_MS, DEFAULT_MAX_IDLE_MS);
            this.peer = peer;
        }

        @Override
        public void processItems(List<AbstractEntry<K, V>> items) {
            Map<K, AbstractEntry<K, V>> map = Maps.newHashMap();
            items.forEach(item -> map.compute(item.key(), (key, oldValue) ->
                  oldValue == null || item.compareTo(oldValue) > 0 ? item : oldValue
                  )
            );
            communicationExecutor.submit(() -> {
                try {
                    unicastMessage(peer, updateMessageSubject, Lists.newArrayList(map.values()));
                } catch (Exception e) {
                    log.warn("broadcast error", e);
                }
            });
        }
    }
}