+public class DistributedDataStore implements DistributedDataStoreInterface, SchemaContextListener,
+ DatastoreContextConfigAdminOverlay.Listener, DOMStoreTreeChangePublisher,
+ DOMDataTreeCommitCohortRegistry, AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DistributedDataStore.class);
+
+ private static final long READY_WAIT_FACTOR = 3;
+
+ private final ActorContext actorContext;
+ private final long waitTillReadyTimeInMillis;
+
+ private AutoCloseable closeable;
+
+ private DatastoreConfigurationMXBeanImpl datastoreConfigMXBean;
+
+ private DatastoreInfoMXBeanImpl datastoreInfoMXBean;
+
+ private final CountDownLatch waitTillReadyCountDownLatch = new CountDownLatch(1);
+
+ private final ClientIdentifier identifier;
+ private final DistributedDataStoreClient client;
+
+ private final TransactionContextFactory txContextFactory;
+
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ public DistributedDataStore(final ActorSystem actorSystem, final ClusterWrapper cluster,
+ final Configuration configuration, final DatastoreContextFactory datastoreContextFactory,
+ final DatastoreSnapshot restoreFromSnapshot) {
+ Preconditions.checkNotNull(actorSystem, "actorSystem should not be null");
+ Preconditions.checkNotNull(cluster, "cluster should not be null");
+ Preconditions.checkNotNull(configuration, "configuration should not be null");
+ Preconditions.checkNotNull(datastoreContextFactory, "datastoreContextFactory should not be null");
+
+ String shardManagerId = ShardManagerIdentifier.builder()
+ .type(datastoreContextFactory.getBaseDatastoreContext().getDataStoreName()).build().toString();
+
+ LOG.info("Creating ShardManager : {}", shardManagerId);
+
+ String shardDispatcher =
+ new Dispatchers(actorSystem.dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard);
+
+ PrimaryShardInfoFutureCache primaryShardInfoCache = new PrimaryShardInfoFutureCache();
+
+ ShardManagerCreator creator = new ShardManagerCreator().cluster(cluster).configuration(configuration)
+ .datastoreContextFactory(datastoreContextFactory)
+ .waitTillReadyCountdownLatch(waitTillReadyCountDownLatch)
+ .primaryShardInfoCache(primaryShardInfoCache)
+ .restoreFromSnapshot(restoreFromSnapshot);
+
+ actorContext = new ActorContext(actorSystem, createShardManager(actorSystem, creator, shardDispatcher,
+ shardManagerId), cluster, configuration, datastoreContextFactory.getBaseDatastoreContext(),
+ primaryShardInfoCache);
+
+ final Props clientProps = DistributedDataStoreClientActor.props(cluster.getCurrentMemberName(),
+ datastoreContextFactory.getBaseDatastoreContext().getDataStoreName(), actorContext);
+ final ActorRef clientActor = actorSystem.actorOf(clientProps);
+ try {
+ client = DistributedDataStoreClientActor.getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ LOG.error("Failed to get actor for {}", clientProps, e);
+ clientActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
+ throw Throwables.propagate(e);
+ }
+
+ identifier = client.getIdentifier();
+ LOG.debug("Distributed data store client {} started", identifier);
+
+ this.waitTillReadyTimeInMillis = actorContext.getDatastoreContext().getShardLeaderElectionTimeout()
+ .duration().toMillis() * READY_WAIT_FACTOR;
+
+ this.txContextFactory = new TransactionContextFactory(actorContext, identifier);
+
+ datastoreConfigMXBean = new DatastoreConfigurationMXBeanImpl(
+ datastoreContextFactory.getBaseDatastoreContext().getDataStoreMXBeanType());
+ datastoreConfigMXBean.setContext(datastoreContextFactory.getBaseDatastoreContext());
+ datastoreConfigMXBean.registerMBean();
+
+ datastoreInfoMXBean = new DatastoreInfoMXBeanImpl(datastoreContextFactory.getBaseDatastoreContext()
+ .getDataStoreMXBeanType(), actorContext);
+ datastoreInfoMXBean.registerMBean();
+ }
+
+ @VisibleForTesting
+ DistributedDataStore(final ActorContext actorContext, final ClientIdentifier identifier) {
+ this.actorContext = Preconditions.checkNotNull(actorContext, "actorContext should not be null");
+ this.client = null;
+ this.identifier = Preconditions.checkNotNull(identifier);
+ this.txContextFactory = new TransactionContextFactory(actorContext, identifier);
+ this.waitTillReadyTimeInMillis = actorContext.getDatastoreContext().getShardLeaderElectionTimeout()
+ .duration().toMillis() * READY_WAIT_FACTOR;
+ }
+
+ public void setCloseable(final AutoCloseable closeable) {
+ this.closeable = closeable;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public <L extends AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>>
+ ListenerRegistration<L> registerChangeListener(
+ final YangInstanceIdentifier path, final L listener,
+ final AsyncDataBroker.DataChangeScope scope) {
+
+ Preconditions.checkNotNull(path, "path should not be null");
+ Preconditions.checkNotNull(listener, "listener should not be null");
+
+ LOG.debug("Registering listener: {} for path: {} scope: {}", listener, path, scope);
+
+ String shardName = actorContext.getShardStrategyFactory().getStrategy(path).findShard(path);
+
+ final DataChangeListenerRegistrationProxy listenerRegistrationProxy =
+ new DataChangeListenerRegistrationProxy(shardName, actorContext, listener);
+ listenerRegistrationProxy.init(path, scope);
+
+ return listenerRegistrationProxy;
+ }