Madan Jampani
Committed by Gerrit Code Review

CachingAsyncConsistentMap: When changes are detected update cache with new value

Change-Id: I51307a8bff953389feeb8928f591151058d49eab
...@@ -51,7 +51,7 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa ...@@ -51,7 +51,7 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa
51 51
52 private final LoadingCache<K, CompletableFuture<Versioned<V>>> cache; 52 private final LoadingCache<K, CompletableFuture<Versioned<V>>> cache;
53 53
54 - private final MapEventListener<K, V> cacheInvalidator; 54 + private final MapEventListener<K, V> cacheUpdater;
55 private final Consumer<Status> statusListener; 55 private final Consumer<Status> statusListener;
56 56
57 /** 57 /**
...@@ -74,7 +74,14 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa ...@@ -74,7 +74,14 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa
74 cache = CacheBuilder.newBuilder() 74 cache = CacheBuilder.newBuilder()
75 .maximumSize(cacheSize) 75 .maximumSize(cacheSize)
76 .build(CacheLoader.from(CachingAsyncConsistentMap.super::get)); 76 .build(CacheLoader.from(CachingAsyncConsistentMap.super::get));
77 - cacheInvalidator = event -> cache.invalidate(event.key()); 77 + cacheUpdater = event -> {
78 + Versioned<V> newValue = event.newValue();
79 + if (newValue == null) {
80 + cache.invalidate(event.key());
81 + } else {
82 + cache.put(event.key(), CompletableFuture.completedFuture(newValue));
83 + }
84 + };
78 statusListener = status -> { 85 statusListener = status -> {
79 log.debug("{} status changed to {}", this.name(), status); 86 log.debug("{} status changed to {}", this.name(), status);
80 // If the status of the underlying map is SUSPENDED or INACTIVE 87 // If the status of the underlying map is SUSPENDED or INACTIVE
...@@ -83,14 +90,14 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa ...@@ -83,14 +90,14 @@ public class CachingAsyncConsistentMap<K, V> extends DelegatingAsyncConsistentMa
83 cache.invalidateAll(); 90 cache.invalidateAll();
84 } 91 }
85 }; 92 };
86 - super.addListener(cacheInvalidator); 93 + super.addListener(cacheUpdater);
87 super.addStatusChangeListener(statusListener); 94 super.addStatusChangeListener(statusListener);
88 } 95 }
89 96
90 @Override 97 @Override
91 public CompletableFuture<Void> destroy() { 98 public CompletableFuture<Void> destroy() {
92 super.removeStatusChangeListener(statusListener); 99 super.removeStatusChangeListener(statusListener);
93 - return super.destroy().thenCompose(v -> removeListener(cacheInvalidator)); 100 + return super.destroy().thenCompose(v -> removeListener(cacheUpdater));
94 } 101 }
95 102
96 @Override 103 @Override
......