+
+ @Override
+ void handleReplayedLocalRequest(final AbstractLocalTransactionRequest<?> request,
+ final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ if (request instanceof CommitLocalTransactionRequest) {
+ replayLocalCommitRequest((CommitLocalTransactionRequest) request, callback, enqueuedTicks);
+ } else if (request instanceof AbortLocalTransactionRequest) {
+ enqueueRequest(abortRequest(), callback, enqueuedTicks);
+ } else {
+ throw new IllegalStateException("Unhandled request " + request);
+ }
+ }
+
+ private void replayLocalCommitRequest(final CommitLocalTransactionRequest request,
+ final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ final DataTreeModification mod = request.getModification();
+ final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
+
+ mod.applyToCursor(new AbstractDataTreeModificationCursor() {
+ @Override
+ public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
+ appendModification(new TransactionWrite(current().node(child), data), optTicks);
+ }
+
+ @Override
+ public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
+ appendModification(new TransactionMerge(current().node(child), data), optTicks);
+ }
+
+ @Override
+ public void delete(final PathArgument child) {
+ appendModification(new TransactionDelete(current().node(child)), optTicks);
+ }
+ });
+
+ enqueueRequest(commitRequest(request.isCoordinated()), callback, enqueuedTicks);
+ }
+
+ @Override
+ void handleReplayedRemoteRequest(final TransactionRequest<?> request,
+ final @Nullable Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ final Consumer<Response<?, ?>> cb = callback != null ? callback : resp -> { };
+ final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
+
+ if (request instanceof ModifyTransactionRequest) {
+ final ModifyTransactionRequest req = (ModifyTransactionRequest) request;
+ for (TransactionModification mod : req.getModifications()) {
+ appendModification(mod, optTicks);
+ }
+
+ final java.util.Optional<PersistenceProtocol> maybeProto = req.getPersistenceProtocol();
+ if (maybeProto.isPresent()) {
+ ensureSealed();
+
+ final TransactionRequest<?> tmp;
+ switch (maybeProto.get()) {
+ case ABORT:
+ tmp = abortRequest();
+ enqueueRequest(tmp, resp -> {
+ completeModify(tmp, resp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ break;
+ case SIMPLE:
+ tmp = commitRequest(false);
+ enqueueRequest(tmp, resp -> {
+ completeModify(tmp, resp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ break;
+ case THREE_PHASE:
+ tmp = commitRequest(true);
+ enqueueRequest(tmp, resp -> {
+ recordSuccessfulRequest(tmp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ break;
+ case READY:
+ //no op
+ break;
+ default:
+ throw new IllegalArgumentException("Unhandled protocol " + maybeProto.get());
+ }
+ }
+ } else if (request instanceof ReadTransactionRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(),
+ ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
+ recordFinishedRequest(resp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ } else if (request instanceof ExistsTransactionRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(),
+ ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), resp -> {
+ recordFinishedRequest(resp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ } else if (request instanceof TransactionPreCommitRequest) {
+ ensureFlushedBuider(optTicks);
+ final TransactionRequest<?> tmp = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
+ localActor());
+ enqueueRequest(tmp, resp -> {
+ recordSuccessfulRequest(tmp);
+ cb.accept(resp);
+ }, enqueuedTicks);
+ } else if (request instanceof TransactionDoCommitRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), callback,
+ enqueuedTicks);
+ } else if (request instanceof TransactionAbortRequest) {
+ ensureFlushedBuider(optTicks);
+ enqueueDoAbort(callback, enqueuedTicks);
+ } else if (request instanceof TransactionPurgeRequest) {
+ enqueuePurge(callback, enqueuedTicks);
+ } else if (request instanceof IncrementTransactionSequenceRequest) {
+ final IncrementTransactionSequenceRequest req = (IncrementTransactionSequenceRequest) request;
+ ensureFlushedBuider(optTicks);
+ enqueueRequest(new IncrementTransactionSequenceRequest(getIdentifier(), nextSequence(), localActor(),
+ snapshotOnly, req.getIncrement()), callback, enqueuedTicks);
+ incrementSequence(req.getIncrement());
+ } else {
+ throw new IllegalArgumentException("Unhandled request {}" + request);
+ }
+ }