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;
import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
+import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
-import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
+import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.reporting.MetricsReporter;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
* 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";
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) {
primaryShardInfoCache = CacheBuilder.newBuilder()
.expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS)
+ .removalListener(this)
.build();
}
return future.transform(new Mapper<Object, PrimaryShardInfo>() {
@Override
public PrimaryShardInfo checkedApply(Object response) throws Exception {
- if(response instanceof PrimaryFound) {
- PrimaryFound found = (PrimaryFound)response;
-
- LOG.debug("Primary found {}", found.getPrimaryPath());
- ActorSelection actorSelection = actorSystem.actorSelection(found.getPrimaryPath());
- PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.<DataTree>absent());
- primaryShardInfoCache.put(shardName, Futures.successful(info));
- return info;
+ if(response instanceof RemotePrimaryShardFound) {
+ LOG.debug("findPrimaryShardAsync received: {}", response);
+ return onPrimaryShardFound(shardName, ((RemotePrimaryShardFound)response).getPrimaryPath(), null);
+ } else if(response instanceof LocalPrimaryShardFound) {
+ LOG.debug("findPrimaryShardAsync received: {}", response);
+ LocalPrimaryShardFound found = (LocalPrimaryShardFound)response;
+ return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getLocalShardDataTree());
} else if(response instanceof NotInitializedException) {
throw (NotInitializedException)response;
} else if(response instanceof PrimaryNotFoundException) {
}, FIND_PRIMARY_FAILURE_TRANSFORMER, getClientDispatcher());
}
+ private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath,
+ DataTree localShardDataTree) {
+ 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;
+ }
+
/**
* Finds a local shard given its shard name and return it's ActorRef
*
}
public void shutdown() {
- shardManager.tell(PoisonPill.getInstance(), null);
- actorSystem.shutdown();
+ shardManager.tell(PoisonPill.getInstance(), ActorRef.noSender());
}
public ClusterWrapper getClusterWrapper() {
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);
+ }
+ }
+ }
}