-public class DistributedDataStore implements DOMStore {
- private final ActorRef shardManager;
-
- public DistributedDataStore(ActorSystem actorSystem, String type) {
- shardManager = actorSystem.actorOf(ShardManager.props(type));
- }
-
- @Override
- public <L extends AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(InstanceIdentifier path, L listener, AsyncDataBroker.DataChangeScope scope) {
- return new ListenerRegistrationProxy();
- }
-
- @Override
- public DOMStoreTransactionChain createTransactionChain() {
- return new TransactionChainProxy();
- }
-
- @Override
- public DOMStoreReadTransaction newReadOnlyTransaction() {
- return new TransactionProxy();
- }
-
- @Override
- public DOMStoreWriteTransaction newWriteOnlyTransaction() {
- return new TransactionProxy();
- }
-
- @Override
- public DOMStoreReadWriteTransaction newReadWriteTransaction() {
- return new TransactionProxy();
- }
+public class DistributedDataStore implements DOMStore, SchemaContextListener, AutoCloseable {
+
+ private static final Logger
+ LOG = LoggerFactory.getLogger(DistributedDataStore.class);
+
+
+ private final String type;
+ private final ActorContext actorContext;
+
+ private SchemaContext schemaContext;
+
+
+
+ /**
+ * Executor used to run FutureTask's
+ *
+ * This is typically used when we need to make a request to an actor and
+ * wait for it's response and the consumer needs to be provided a Future.
+ *
+ * FIXME : Make the thread pool configurable
+ */
+ private final ExecutorService executor =
+ Executors.newFixedThreadPool(10);
+
+ public DistributedDataStore(ActorSystem actorSystem, String type) {
+ this(new ActorContext(actorSystem, actorSystem.actorOf(ShardManager.props(type), "shardmanager-" + type)), type);
+ }
+
+ public DistributedDataStore(ActorContext actorContext, String type) {
+ this.type = type;
+ this.actorContext = actorContext;
+ }
+
+
+ @Override
+ public <L extends AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>> ListenerRegistration<L> registerChangeListener(
+ InstanceIdentifier path, L listener,
+ AsyncDataBroker.DataChangeScope scope) {
+
+ ActorRef dataChangeListenerActor = actorContext.getActorSystem().actorOf(
+ DataChangeListener.props(listener));
+
+ Object result = actorContext.executeShardOperation(Shard.DEFAULT_NAME,
+ new RegisterChangeListener(path, dataChangeListenerActor.path(),
+ AsyncDataBroker.DataChangeScope.BASE),
+ ActorContext.ASK_DURATION
+ );
+
+ RegisterChangeListenerReply reply = (RegisterChangeListenerReply) result;
+ return new DataChangeListenerRegistrationProxy(actorContext.actorSelection(reply.getListenerRegistrationPath()), listener, dataChangeListenerActor);
+ }
+
+
+
+ @Override
+ public DOMStoreTransactionChain createTransactionChain() {
+ return new TransactionChainProxy(actorContext, executor, schemaContext);
+ }
+
+ @Override
+ public DOMStoreReadTransaction newReadOnlyTransaction() {
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_ONLY,
+ executor, schemaContext);
+ }
+
+ @Override
+ public DOMStoreWriteTransaction newWriteOnlyTransaction() {
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.WRITE_ONLY,
+ executor, schemaContext);
+ }
+
+ @Override
+ public DOMStoreReadWriteTransaction newReadWriteTransaction() {
+ return new TransactionProxy(actorContext, TransactionProxy.TransactionType.READ_WRITE,
+ executor, schemaContext);
+ }
+
+ @Override public void onGlobalContextUpdated(SchemaContext schemaContext) {
+ this.schemaContext = schemaContext;
+ actorContext.getShardManager().tell(
+ new UpdateSchemaContext(schemaContext), null);
+ }
+
+ @Override public void close() throws Exception {
+ actorContext.shutdown();
+
+ }