@Test public void testCompute() 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.REMOVE, KEY1, VALUE1)); listener.event(new EventuallyConsistentMapEvent<>( MAP_NAME, EventuallyConsistentMapEvent.Type.PUT, KEY2, VALUE2)); replay(listener); ecMap.addListener(listener); expectPeerMessage(clusterCommunicator); ecMap.compute(KEY1, (k, v) -> VALUE1); assertEquals(VALUE1, ecMap.get(KEY1)); expectSpecificMulticastMessage(generateRemoveMessage(KEY1, clockService.peekAtNextTimestamp()), UPDATE_MESSAGE_SUBJECT, clusterCommunicator); ecMap.compute(KEY1, (k, v) -> null); assertNull(ecMap.get(KEY1)); verify(clusterCommunicator); expectSpecificMulticastMessage(generateRemoveMessage(KEY1, clockService.peekAtNextTimestamp()), UPDATE_MESSAGE_SUBJECT, clusterCommunicator); ecMap.compute(KEY1, (k, v) -> null); assertNull(ecMap.get(KEY1)); verify(clusterCommunicator); expectPeerMessage(clusterCommunicator); ecMap.compute(KEY2, (k, v) -> VALUE2); clockService.turnBackTime(); reset(clusterCommunicator); replay(clusterCommunicator); ecMap.compute(KEY2, (k, v) -> null); verify(clusterCommunicator); verify(listener); }
@Override public V compute(K key, BiFunction<K, V, V> recomputeFunction) { checkState(!destroyed, destroyedMessage); checkNotNull(key, ERROR_NULL_KEY); checkNotNull(recomputeFunction, "Recompute function cannot be null"); AtomicBoolean updated = new AtomicBoolean(false); AtomicReference<MapValue<V>> previousValue = new AtomicReference<>(); MapValue<V> computedValue = items.compute(serializer.copy(key), (k, mv) -> { previousValue.set(mv); V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get()); MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue)); if (mv == null || newValue.isNewerThan(mv)) { updated.set(true); return serializer.copy(newValue); } else { return mv; } }); if (updated.get()) { notifyPeers(new UpdateEntry<>(key, computedValue), peerUpdateFunction.apply(key, computedValue.get())); EventuallyConsistentMapEvent.Type updateType = computedValue.isTombstone() ? REMOVE : PUT; V value = computedValue.isTombstone() ? previousValue.get() == null ? null : previousValue.get().get() : computedValue.get(); if (value != null) { notifyListeners(new EventuallyConsistentMapEvent<>(mapName, updateType, key, value)); } } return computedValue.get(); }
EventuallyConsistentMapImpl implements EventuallyConsistentMap<K, V> { @Override public V compute(K key, BiFunction<K, V, V> recomputeFunction) { checkState(!destroyed, destroyedMessage); checkNotNull(key, ERROR_NULL_KEY); checkNotNull(recomputeFunction, "Recompute function cannot be null"); AtomicBoolean updated = new AtomicBoolean(false); AtomicReference<MapValue<V>> previousValue = new AtomicReference<>(); MapValue<V> computedValue = items.compute(serializer.copy(key), (k, mv) -> { previousValue.set(mv); V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get()); MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue)); if (mv == null || newValue.isNewerThan(mv)) { updated.set(true); return serializer.copy(newValue); } else { return mv; } }); if (updated.get()) { notifyPeers(new UpdateEntry<>(key, computedValue), peerUpdateFunction.apply(key, computedValue.get())); EventuallyConsistentMapEvent.Type updateType = computedValue.isTombstone() ? REMOVE : PUT; V value = computedValue.isTombstone() ? previousValue.get() == null ? null : previousValue.get().get() : computedValue.get(); if (value != null) { notifyListeners(new EventuallyConsistentMapEvent<>(mapName, updateType, key, value)); } } return computedValue.get(); } }
EventuallyConsistentMapImpl implements EventuallyConsistentMap<K, V> { @Override public V compute(K key, BiFunction<K, V, V> recomputeFunction) { checkState(!destroyed, destroyedMessage); checkNotNull(key, ERROR_NULL_KEY); checkNotNull(recomputeFunction, "Recompute function cannot be null"); AtomicBoolean updated = new AtomicBoolean(false); AtomicReference<MapValue<V>> previousValue = new AtomicReference<>(); MapValue<V> computedValue = items.compute(serializer.copy(key), (k, mv) -> { previousValue.set(mv); V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get()); MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue)); if (mv == null || newValue.isNewerThan(mv)) { updated.set(true); return serializer.copy(newValue); } else { return mv; } }); if (updated.get()) { notifyPeers(new UpdateEntry<>(key, computedValue), peerUpdateFunction.apply(key, computedValue.get())); EventuallyConsistentMapEvent.Type updateType = computedValue.isTombstone() ? REMOVE : PUT; V value = computedValue.isTombstone() ? previousValue.get() == null ? null : previousValue.get().get() : computedValue.get(); if (value != null) { notifyListeners(new EventuallyConsistentMapEvent<>(mapName, updateType, key, value)); } } return computedValue.get(); } 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 V compute(K key, BiFunction<K, V, V> recomputeFunction) { checkState(!destroyed, destroyedMessage); checkNotNull(key, ERROR_NULL_KEY); checkNotNull(recomputeFunction, "Recompute function cannot be null"); AtomicBoolean updated = new AtomicBoolean(false); AtomicReference<MapValue<V>> previousValue = new AtomicReference<>(); MapValue<V> computedValue = items.compute(serializer.copy(key), (k, mv) -> { previousValue.set(mv); V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get()); MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue)); if (mv == null || newValue.isNewerThan(mv)) { updated.set(true); return serializer.copy(newValue); } else { return mv; } }); if (updated.get()) { notifyPeers(new UpdateEntry<>(key, computedValue), peerUpdateFunction.apply(key, computedValue.get())); EventuallyConsistentMapEvent.Type updateType = computedValue.isTombstone() ? REMOVE : PUT; V value = computedValue.isTombstone() ? previousValue.get() == null ? null : previousValue.get().get() : computedValue.get(); if (value != null) { notifyListeners(new EventuallyConsistentMapEvent<>(mapName, updateType, key, value)); } } return computedValue.get(); } 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 V compute(K key, BiFunction<K, V, V> recomputeFunction) { checkState(!destroyed, destroyedMessage); checkNotNull(key, ERROR_NULL_KEY); checkNotNull(recomputeFunction, "Recompute function cannot be null"); AtomicBoolean updated = new AtomicBoolean(false); AtomicReference<MapValue<V>> previousValue = new AtomicReference<>(); MapValue<V> computedValue = items.compute(serializer.copy(key), (k, mv) -> { previousValue.set(mv); V newRawValue = recomputeFunction.apply(key, mv == null ? null : mv.get()); MapValue<V> newValue = new MapValue<>(newRawValue, timestampProvider.apply(key, newRawValue)); if (mv == null || newValue.isNewerThan(mv)) { updated.set(true); return serializer.copy(newValue); } else { return mv; } }); if (updated.get()) { notifyPeers(new UpdateEntry<>(key, computedValue), peerUpdateFunction.apply(key, computedValue.get())); EventuallyConsistentMapEvent.Type updateType = computedValue.isTombstone() ? REMOVE : PUT; V value = computedValue.isTombstone() ? previousValue.get() == null ? null : previousValue.get().get() : computedValue.get(); if (value != null) { notifyListeners(new EventuallyConsistentMapEvent<>(mapName, updateType, key, value)); } } return computedValue.get(); } 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();  }