@Test public void testPut() throws Exception { 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); expectSpecificMulticastMessage(generatePutMessage(KEY1, VALUE1, clockService .peekAtNextTimestamp()), UPDATE_MESSAGE_SUBJECT, clusterCommunicator); assertNull(ecMap.get(KEY1)); ecMap.put(KEY1, VALUE1); assertEquals(VALUE1, ecMap.get(KEY1)); verify(clusterCommunicator); expectSpecificMulticastMessage(generatePutMessage( KEY1, VALUE2, clockService.peekAtNextTimestamp()), UPDATE_MESSAGE_SUBJECT, clusterCommunicator); ecMap.put(KEY1, VALUE2); assertEquals(VALUE2, ecMap.get(KEY1)); verify(clusterCommunicator); reset(clusterCommunicator); replay(clusterCommunicator); clockService.turnBackTime(); ecMap.put(KEY1, VALUE1); assertEquals(VALUE2, ecMap.get(KEY1)); verify(clusterCommunicator); verify(listener); }
@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)); } }
EventuallyConsistentMapImpl implements EventuallyConsistentMap<K, V> { @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)); } } }
EventuallyConsistentMapImpl implements EventuallyConsistentMap<K, V> { @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)); } } EventuallyConsistentMapImpl(String mapName,
                                ClusterService clusterService,
                                ClusterCommunicationService clusterCommunicator,
                                KryoNamespace ns,
                                BiFunction<K, V, Timestamp> timestampProvider,
                                BiFunction<K, V, Collection<NodeId>> peerUpdateFunction,
                                ExecutorService eventExecutor,
                                ExecutorService communicationExecutor,
                                ScheduledExecutorService backgroundExecutor,
                                boolean tombstonesDisabled,
                                long antiEntropyPeriod,
                                TimeUnit antiEntropyTimeUnit,
                                boolean convergeFaster,
                                boolean persistent,
                                PersistenceService persistenceService); }
EventuallyConsistentMapImpl implements EventuallyConsistentMap<K, V> { @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)); } } EventuallyConsistentMapImpl(String mapName,
                                ClusterService clusterService,
                                ClusterCommunicationService clusterCommunicator,
                                KryoNamespace ns,
                                BiFunction<K, V, Timestamp> timestampProvider,
                                BiFunction<K, V, Collection<NodeId>> peerUpdateFunction,
                                ExecutorService eventExecutor,
                                ExecutorService communicationExecutor,
                                ScheduledExecutorService backgroundExecutor,
                                boolean tombstonesDisabled,
                                long antiEntropyPeriod,
                                TimeUnit antiEntropyTimeUnit,
                                boolean convergeFaster,
                                boolean persistent,
                                PersistenceService persistenceService); @Override String name(); @Override int size(); @Override boolean isEmpty(); @Override boolean containsKey(K key); @Override boolean containsValue(V value); @Override V get(K key); @Override void put(K key, V value); @Override V remove(K key); @Override void remove(K key, V value); @Override V compute(K key, BiFunction<K, V, V> recomputeFunction); @Override void putAll(Map<? extends K, ? extends V> m); @Override void clear(); @Override Set<K> keySet(); @Override Collection<V> values(); @Override Set<Map.Entry<K, V>> entrySet(); @Override void addListener(EventuallyConsistentMapListener<K, V> listener); @Override void removeListener(EventuallyConsistentMapListener<K, V> listener); @Override CompletableFuture<Void> destroy(); }
EventuallyConsistentMapImpl implements EventuallyConsistentMap<K, V> { @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)); } } EventuallyConsistentMapImpl(String mapName,
                                ClusterService clusterService,
                                ClusterCommunicationService clusterCommunicator,
                                KryoNamespace ns,
                                BiFunction<K, V, Timestamp> timestampProvider,
                                BiFunction<K, V, Collection<NodeId>> peerUpdateFunction,
                                ExecutorService eventExecutor,
                                ExecutorService communicationExecutor,
                                ScheduledExecutorService backgroundExecutor,
                                boolean tombstonesDisabled,
                                long antiEntropyPeriod,
                                TimeUnit antiEntropyTimeUnit,
                                boolean convergeFaster,
                                boolean persistent,
                                PersistenceService persistenceService); @Override String name(); @Override int size(); @Override boolean isEmpty(); @Override boolean containsKey(K key); @Override boolean containsValue(V value); @Override V get(K key); @Override void put(K key, V value); @Override V remove(K key); @Override void remove(K key, V value); @Override V compute(K key, BiFunction<K, V, V> recomputeFunction); @Override void putAll(Map<? extends K, ? extends V> m); @Override void clear(); @Override Set<K> keySet(); @Override Collection<V> values(); @Override Set<Map.Entry<K, V>> entrySet(); @Override void addListener(EventuallyConsistentMapListener<K, V> listener); @Override void removeListener(EventuallyConsistentMapListener<K, V> listener); @Override CompletableFuture<Void> destroy();  }