import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfo;
-import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfoMBean;
import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
import org.opendaylight.controller.cluster.datastore.messages.ActorNotInitialized;
import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
+import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
import org.opendaylight.yangtools.yang.model.api.ModuleIdentifier;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.slf4j.Logger;
private final Configuration configuration;
- private ShardManagerInfoMBean mBean;
+ private final String shardDispatcherPath;
- private final DatastoreContext datastoreContext;
+ private ShardManagerInfo mBean;
+
+ private DatastoreContext datastoreContext;
private Collection<String> knownModules = Collections.emptySet();
this.datastoreContext = datastoreContext;
this.dataPersistenceProvider = createDataPersistenceProvider(datastoreContext.isPersistent());
this.type = datastoreContext.getDataStoreType();
+ this.shardDispatcherPath =
+ new Dispatchers(context().system().dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard);
// Subscribe this actor to cluster member events
cluster.subscribeToMemberEvents(getSelf());
return Props.create(new ShardManagerCreator(cluster, configuration, datastoreContext));
}
+ @Override
+ public void postStop() {
+ LOG.info("Stopping ShardManager");
+
+ mBean.unregisterMBean();
+ }
+
@Override
public void handleCommand(Object message) throws Exception {
if (message.getClass().equals(FindPrimary.SERIALIZABLE_CLASS)) {
memberRemoved((ClusterEvent.MemberRemoved) message);
} else if(message instanceof ClusterEvent.UnreachableMember) {
ignoreMessage(message);
+ } else if(message instanceof DatastoreContext) {
+ onDatastoreContext((DatastoreContext)message);
} else{
unknownMessage(message);
}
}
}
+ private void onDatastoreContext(DatastoreContext context) {
+ datastoreContext = context;
+ for (ShardInformation info : localShards.values()) {
+ if (info.getActor() != null) {
+ info.getActor().tell(datastoreContext, getSelf());
+ }
+ }
+ }
+
/**
* Notifies all the local shards of a change in the schema context
*
if(newModules.containsAll(knownModules)) {
- LOG.info("New SchemaContext has a super set of current knownModules - persisting info");
+ LOG.debug("New SchemaContext has a super set of current knownModules - persisting info");
knownModules = ImmutableSet.copyOf(newModules);
@Override
public void apply(SchemaContextModules param) throws Exception {
- LOG.info("Sending new SchemaContext to Shards");
+ LOG.debug("Sending new SchemaContext to Shards");
for (ShardInformation info : localShards.values()) {
if (info.getActor() == null) {
info.setActor(getContext().actorOf(Shard.props(info.getShardId(),
- info.getPeerAddresses(), datastoreContext, schemaContext),
- info.getShardId().toString()));
+ info.getPeerAddresses(), datastoreContext, schemaContext)
+ .withDispatcher(shardDispatcherPath), info.getShardId().toString()));
} else {
info.getActor().tell(message, getSelf());
}
});
} else {
- LOG.info("Rejecting schema context update because it is not a super set of previously known modules");
+ LOG.debug("Rejecting schema context update - not a super set of previously known modules:\nUPDATE: {}\nKNOWN: {}",
+ newModules, knownModules);
}
}