CDS: Implement front-end support for local transactions

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / utils / ActorContext.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java
index 18a8798c4755d80a982550ffc3fee42c9d1f9d0f..ad05a1ca71001285a8aed1f5afe16b1f511f3e96 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java
@@ -28,8 +28,13 @@ import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
 import com.google.common.util.concurrent.RateLimiter;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.concurrent.GuardedBy;
 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
 import org.opendaylight.controller.cluster.datastore.Configuration;
@@ -65,7 +70,7 @@ import scala.concurrent.duration.FiniteDuration;
  * easily. An ActorContext can be freely passed around to local object instances
  * but should not be passed to actors especially remote actors
  */
-public class ActorContext {
+public class ActorContext implements RemovalListener<String, Future<PrimaryShardInfo>> {
     private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class);
     private static final String DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store";
     private static final String METRIC_RATE = "rate";
@@ -104,6 +109,8 @@ public class ActorContext {
     private volatile SchemaContext schemaContext;
     private volatile boolean updated;
     private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN).getMetricsRegistry();
+    @GuardedBy("shardInfoListeners")
+    private final Collection<ShardInfoListenerRegistration<?>> shardInfoListeners = new ArrayList<>();
 
     public ActorContext(ActorSystem actorSystem, ActorRef shardManager,
             ClusterWrapper clusterWrapper, Configuration configuration) {
@@ -146,6 +153,7 @@ public class ActorContext {
 
         primaryShardInfoCache = CacheBuilder.newBuilder()
                 .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS)
+                .removalListener(this)
                 .build();
     }
 
@@ -236,6 +244,12 @@ public class ActorContext {
         ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath);
         PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.fromNullable(localShardDataTree));
         primaryShardInfoCache.put(shardName, Futures.successful(info));
+
+        synchronized (shardInfoListeners) {
+            for (ShardInfoListenerRegistration<?> reg : shardInfoListeners) {
+                reg.getInstance().onShardInfoUpdated(shardName, info);
+            }
+        }
         return info;
     }
 
@@ -566,4 +580,28 @@ public class ActorContext {
     Cache<String, Future<PrimaryShardInfo>> getPrimaryShardInfoCache() {
         return primaryShardInfoCache;
     }
+
+    public <T extends ShardInfoListener> ShardInfoListenerRegistration<T> registerShardInfoListener(final T listener) {
+        final ShardInfoListenerRegistration<T> reg = new ShardInfoListenerRegistration<T>(listener, this);
+
+        synchronized (shardInfoListeners) {
+            shardInfoListeners.add(reg);
+        }
+        return reg;
+    }
+
+    protected void removeShardInfoListener(final ShardInfoListenerRegistration<?> registration) {
+        synchronized (shardInfoListeners) {
+            shardInfoListeners.remove(registration);
+        }
+    }
+
+    @Override
+    public void onRemoval(final RemovalNotification<String, Future<PrimaryShardInfo>> notification) {
+        synchronized (shardInfoListeners) {
+            for (ShardInfoListenerRegistration<?> reg : shardInfoListeners) {
+                reg.getInstance().onShardInfoUpdated(notification.getKey(), null);
+            }
+        }
+    }
 }