+ private final Map<Modification, DOMStoreThreePhaseCommitCohort>
+ modificationToCohort = new HashMap<>();
+
+ private final LoggingAdapter log =
+ Logging.getLogger(getContext().system(), this);
+
+ // By default persistent will be true and can be turned off using the system
+ // property persistent
+ private final boolean persistent;
+
+ private Shard(String name) {
+
+ String setting = System.getProperty("shard.persistent");
+ this.persistent = !"false".equals(setting);
+
+ log.info("Creating shard : {} persistent : {}", name , persistent);
+
+ store = new InMemoryDOMDataStore(name, storeExecutor);
+ }
+
+ public static Props props(final String name) {
+ return Props.create(new Creator<Shard>() {
+
+ @Override
+ public Shard create() throws Exception {
+ return new Shard(name);
+ }
+
+ });
+ }
+
+
+ @Override
+ public void onReceive(Object message) throws Exception {
+ log.debug("Received message {}", message);
+
+ if (message instanceof CreateTransactionChain) {
+ createTransactionChain();
+ } else if (message instanceof RegisterChangeListener) {
+ registerChangeListener((RegisterChangeListener) message);
+ } else if (message instanceof UpdateSchemaContext) {
+ updateSchemaContext((UpdateSchemaContext) message);
+ } else if (message instanceof ForwardedCommitTransaction) {
+ handleForwardedCommit((ForwardedCommitTransaction) message);
+ } else if (message instanceof Persistent) {
+ commit((Modification) ((Persistent) message).payload());
+ } else if (message instanceof CreateTransaction) {
+ createTransaction((CreateTransaction) message);
+ } else if(message instanceof NonPersistent){
+ commit((Modification) ((NonPersistent) message).payload());
+ }
+ }
+
+ private void createTransaction(CreateTransaction createTransaction) {
+ DOMStoreReadWriteTransaction transaction =
+ store.newReadWriteTransaction();
+ ActorRef transactionActor = getContext().actorOf(
+ ShardTransaction.props(transaction, getSelf()), "shard-" + createTransaction.getTransactionId());
+ getSender()
+ .tell(new CreateTransactionReply(transactionActor.path(), createTransaction.getTransactionId()),
+ getSelf());
+ }
+
+ private void commit(Modification modification) {
+ DOMStoreThreePhaseCommitCohort cohort =
+ modificationToCohort.remove(modification);
+ if (cohort == null) {
+ log.error(
+ "Could not find cohort for modification : " + modification);
+ return;
+ }
+ final ListenableFuture<Void> future = cohort.commit();
+ final ActorRef sender = getSender();
+ final ActorRef self = getSelf();
+ future.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ future.get();
+ sender.tell(new CommitTransactionReply(), self);
+ } catch (InterruptedException | ExecutionException e) {
+ // FIXME : Handle this properly
+ log.error(e, "An exception happened when committing");
+ }
+ }
+ }, getContext().dispatcher());
+ }
+
+ private void handleForwardedCommit(ForwardedCommitTransaction message) {
+ modificationToCohort
+ .put(message.getModification(), message.getCohort());
+ if(persistent) {
+ getSelf().forward(Persistent.create(message.getModification()),
+ getContext());
+ } else {
+ getSelf().forward(NonPersistent.create(message.getModification()),
+ getContext());
+ }
+ }
+
+ private void updateSchemaContext(UpdateSchemaContext message) {
+ store.onGlobalContextUpdated(message.getSchemaContext());
+ }
+
+ private void registerChangeListener(
+ RegisterChangeListener registerChangeListener) {
+
+ ActorSelection dataChangeListenerPath = getContext()
+ .system().actorSelection(registerChangeListener.getDataChangeListenerPath());
+
+ AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>
+ listener = new DataChangeListenerProxy(dataChangeListenerPath);
+
+ org.opendaylight.yangtools.concepts.ListenerRegistration<AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>>
+ registration =
+ store.registerChangeListener(registerChangeListener.getPath(),
+ listener, registerChangeListener.getScope());
+ ActorRef listenerRegistration =
+ getContext().actorOf(
+ DataChangeListenerRegistration.props(registration));
+ getSender()
+ .tell(new RegisterChangeListenerReply(listenerRegistration.path()),
+ getSelf());
+ }
+
+ private void createTransactionChain() {
+ DOMStoreTransactionChain chain = store.createTransactionChain();
+ ActorRef transactionChain =
+ getContext().actorOf(ShardTransactionChain.props(chain));
+ getSender()
+ .tell(new CreateTransactionChainReply(transactionChain.path()),
+ getSelf());