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

        MapValue<V> newValue = new MapValue<>(value, timestampProvider.apply(key, value));
        if (putInternal(key, newValue)) {
            notifyPeers(new UpdateEntry<>(key, newValue), peerUpdateFunction.apply(key, value));
            notifyListeners(new EventuallyConsistentMapEvent<>(mapName, PUT, key, value));
        }
    }
@Test
    public void testPut() throws Exception {
        // Set up expectations of external events to be sent to listeners during
        // the test. These don't use timestamps so we can set them all up at once.
        EventuallyConsistentMapListener<String, String> listener
                = getListener();
        listener.event(new EventuallyConsistentMapEvent<>(
                MAP_NAME, EventuallyConsistentMapEvent.Type.PUT, KEY1, VALUE1));
        listener.event(new EventuallyConsistentMapEvent<>(
                MAP_NAME, EventuallyConsistentMapEvent.Type.PUT, KEY1, VALUE2));
        replay(listener);

        ecMap.addListener(listener);

        // Set up expected internal message to be broadcast to peers on first put
        expectSpecificMulticastMessage(generatePutMessage(KEY1, VALUE1, clockService
                .peekAtNextTimestamp()), UPDATE_MESSAGE_SUBJECT, clusterCommunicator);

        // Put first value
        assertNull(ecMap.get(KEY1));
        ecMap.put(KEY1, VALUE1);
        assertEquals(VALUE1, ecMap.get(KEY1));

        verify(clusterCommunicator);

        // Set up expected internal message to be broadcast to peers on second put
        expectSpecificMulticastMessage(generatePutMessage(
                KEY1, VALUE2, clockService.peekAtNextTimestamp()), UPDATE_MESSAGE_SUBJECT, clusterCommunicator);

        // Update same key to a new value
        ecMap.put(KEY1, VALUE2);
        assertEquals(VALUE2, ecMap.get(KEY1));

        verify(clusterCommunicator);

        // Do a put with a older timestamp than the value already there.
        // The map data should not be changed and no notifications should be sent.
        reset(clusterCommunicator);
        replay(clusterCommunicator);

        clockService.turnBackTime();
        ecMap.put(KEY1, VALUE1);
        // Value should not have changed.
        assertEquals(VALUE2, ecMap.get(KEY1));

        verify(clusterCommunicator);

        // Check that our listener received the correct events during the test
        verify(listener);
    }