X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FAbstractDataStore.java;h=477e05bcc6ecb7c92d1bda3e22cd28c52def06fd;hp=01f2ceb544293477835f3fbe62c55092fd3c012d;hb=08269ac33356663d75f0df8cc54936eac5553e7b;hpb=61791b72e0137609d15c18efc64d227b6a4006ec diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java index 01f2ceb544..477e05bcc6 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataStore.java @@ -28,10 +28,12 @@ import org.opendaylight.controller.cluster.datastore.jmx.mbeans.DatastoreInfoMXB import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot; import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManagerCreator; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; +import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils; import org.opendaylight.controller.cluster.datastore.utils.Dispatchers; import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker; import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener; +import org.opendaylight.controller.md.sal.dom.api.ClusteredDOMDataTreeChangeListener; import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener; import org.opendaylight.controller.sal.core.spi.data.DOMStoreTreeChangePublisher; import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort; @@ -94,7 +96,8 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface .datastoreContextFactory(datastoreContextFactory) .waitTillReadyCountDownLatch(waitTillReadyCountDownLatch) .primaryShardInfoCache(primaryShardInfoCache) - .restoreFromSnapshot(restoreFromSnapshot); + .restoreFromSnapshot(restoreFromSnapshot) + .distributedDataStore(this); actorContext = new ActorContext(actorSystem, createShardManager(actorSystem, creator, shardDispatcher, shardManagerId), cluster, configuration, datastoreContextFactory.getBaseDatastoreContext(), @@ -136,6 +139,16 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface .duration().toMillis() * READY_WAIT_FACTOR; } + @VisibleForTesting + protected AbstractDataStore(final ActorContext actorContext, final ClientIdentifier identifier, + final DataStoreClient clientActor) { + this.actorContext = Preconditions.checkNotNull(actorContext, "actorContext should not be null"); + this.client = clientActor; + this.identifier = Preconditions.checkNotNull(identifier); + this.waitTillReadyTimeInMillis = actorContext.getDatastoreContext().getShardLeaderElectionTimeout() + .duration().toMillis() * READY_WAIT_FACTOR; + } + protected final DataStoreClient getClient() { return client; } @@ -270,8 +283,7 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface for (int i = 0; i < 100; i++) { try { - return actorSystem.actorOf(creator.props().withDispatcher(shardDispatcher).withMailbox( - ActorContext.BOUNDED_MAILBOX), shardManagerId); + return actorSystem.actorOf(creator.props().withDispatcher(shardDispatcher), shardManagerId); } catch (Exception e) { lastException = e; Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); @@ -303,10 +315,28 @@ public abstract class AbstractDataStore implements DistributedDataStoreInterface delegate,shardLookup, shardName, insideShard); final DataTreeChangeListenerProxy listenerRegistrationProxy = - new DataTreeChangeListenerProxy<>(actorContext, delegate::onDataTreeChanged, insideShard); + new DataTreeChangeListenerProxy<>(actorContext, + // wrap this in the ClusteredDOMDataTreeChangeLister interface + // since we always want clustered registration + (ClusteredDOMDataTreeChangeListener) delegate::onDataTreeChanged, insideShard); listenerRegistrationProxy.init(shardName); return (ListenerRegistration) listenerRegistrationProxy; } + @SuppressWarnings("unchecked") + public ListenerRegistration registerShardConfigListener( + final YangInstanceIdentifier internalPath, + final DOMDataTreeChangeListener delegate) { + Preconditions.checkNotNull(delegate, "delegate should not be null"); + + LOG.debug("Registering a listener for the configuration shard: {}", internalPath); + + final DataTreeChangeListenerProxy proxy = + new DataTreeChangeListenerProxy<>(actorContext, delegate, internalPath); + proxy.init(ClusterUtils.PREFIX_CONFIG_SHARD_ID); + + return (ListenerRegistration) proxy; + } + }