+ ActorRef transactionChain = getContext().actorOf(
+ ShardTransactionChain.props(chain, schemaContext, datastoreContext, shardMBean));
+ getSender().tell(new CreateTransactionChainReply(transactionChain.path()).toSerializable(),
+ getSelf());
+ }
+
+ @Override protected void applyState(ActorRef clientActor, String identifier,
+ Object data) {
+
+ if (data instanceof CompositeModificationPayload) {
+ Object modification =
+ ((CompositeModificationPayload) data).getModification();
+
+ if (modification != null) {
+ commit(clientActor, modification);
+ } else {
+ LOG.error(
+ "modification is null - this is very unexpected, clientActor = {}, identifier = {}",
+ identifier, clientActor.path().toString());
+ }
+
+ } else {
+ LOG.error("Unknown state received {}", data);
+ }
+
+ // Update stats
+ ReplicatedLogEntry lastLogEntry = getLastLogEntry();
+
+ if (lastLogEntry != null) {
+ shardMBean.setLastLogIndex(lastLogEntry.getIndex());
+ shardMBean.setLastLogTerm(lastLogEntry.getTerm());
+ }
+
+ shardMBean.setCommitIndex(getCommitIndex());
+ shardMBean.setLastApplied(getLastApplied());
+
+ }
+
+ @Override protected void createSnapshot() {
+ if (createSnapshotTransaction == null) {
+
+ // Create a transaction. We are really going to treat the transaction as a worker
+ // so that this actor does not get block building the snapshot
+ createSnapshotTransaction = createTransaction(
+ TransactionProxy.TransactionType.READ_ONLY.ordinal(),
+ "createSnapshot");
+
+ createSnapshotTransaction.tell(
+ new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self());
+
+ }
+ }
+
+ @VisibleForTesting @Override protected void applySnapshot(ByteString snapshot) {
+ // Since this will be done only on Recovery or when this actor is a Follower
+ // we can safely commit everything in here. We not need to worry about event notifications
+ // as they would have already been disabled on the follower
+ try {
+ DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
+ NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(snapshot);
+ NormalizedNode<?, ?> node = new NormalizedNodeToNodeCodec(schemaContext)
+ .decode(YangInstanceIdentifier.builder().build(), serializedNode);
+
+ // delete everything first
+ transaction.delete(YangInstanceIdentifier.builder().build());
+
+ // Add everything from the remote node back
+ transaction.write(YangInstanceIdentifier.builder().build(), node);
+ syncCommitTransaction(transaction);
+ } catch (InvalidProtocolBufferException | InterruptedException | ExecutionException e) {
+ LOG.error(e, "An exception occurred when applying snapshot");
+ }
+ }
+
+ @Override protected void onStateChanged() {
+ for (ActorSelection dataChangeListener : dataChangeListeners) {
+ dataChangeListener
+ .tell(new EnableNotification(isLeader()), getSelf());
+ }
+
+ if (getLeaderId() != null) {
+ shardMBean.setLeader(getLeaderId());
+ }
+
+ shardMBean.setRaftState(getRaftState().name());
+ shardMBean.setCurrentTerm(getCurrentTerm());
+ }
+
+ @Override public String persistenceId() {
+ return this.name.toString();