+ @Override
+ void handleForwardedLocalRequest(final AbstractLocalTransactionRequest<?> request,
+ final Consumer<Response<?, ?>> callback) {
+ if (request instanceof CommitLocalTransactionRequest) {
+ replayLocalCommitRequest((CommitLocalTransactionRequest) request, callback);
+ } else if (request instanceof AbortLocalTransactionRequest) {
+ sendRequest(abortRequest(), callback);
+ } else {
+ throw new IllegalStateException("Unhandled request " + request);
+ }
+ }
+
+ private void replayLocalCommitRequest(final CommitLocalTransactionRequest request,
+ final Consumer<Response<?, ?>> callback) {
+ final DataTreeModification mod = request.getModification();
+ mod.applyToCursor(new AbstractDataTreeModificationCursor() {
+ @Override
+ public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
+ doWrite(current().node(child), data);
+ }
+
+ @Override
+ public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
+ doMerge(current().node(child), data);
+ }
+
+ @Override
+ public void delete(final PathArgument child) {
+ doDelete(current().node(child));
+ }
+ });
+
+ sendRequest(commitRequest(request.isCoordinated()), callback);
+ }
+