import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
-import com.google.common.base.Verify;
import com.google.common.util.concurrent.CheckedFuture;
import com.google.common.util.concurrent.Futures;
import java.util.function.Consumer;
import javax.annotation.concurrent.NotThreadSafe;
import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess;
import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
-import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
-import org.opendaylight.controller.cluster.access.commands.TransactionDelete;
-import org.opendaylight.controller.cluster.access.commands.TransactionMerge;
-import org.opendaylight.controller.cluster.access.commands.TransactionModification;
+import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ReadTransactionSuccess;
+import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
-import org.opendaylight.controller.cluster.access.commands.TransactionWrite;
-import org.opendaylight.controller.cluster.access.concepts.RequestException;
import org.opendaylight.controller.cluster.access.concepts.Response;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.util.AbstractDataTreeModificationCursor;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeModification;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeSnapshot;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModificationCursor;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
* @author Robert Varga
*/
@NotThreadSafe
-final class LocalProxyTransaction extends AbstractProxyTransaction {
+abstract class LocalProxyTransaction extends AbstractProxyTransaction {
private static final Logger LOG = LoggerFactory.getLogger(LocalProxyTransaction.class);
private final TransactionIdentifier identifier;
- private CursorAwareDataTreeModification modification;
- private CursorAwareDataTreeSnapshot sealedModification;
-
- LocalProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier,
- final CursorAwareDataTreeModification modification) {
+ LocalProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier) {
super(parent);
this.identifier = Preconditions.checkNotNull(identifier);
- this.modification = Preconditions.checkNotNull(modification);
}
@Override
- public TransactionIdentifier getIdentifier() {
+ public final TransactionIdentifier getIdentifier() {
return identifier;
}
- @Override
- void doDelete(final YangInstanceIdentifier path) {
- modification.delete(path);
- }
-
- @Override
- void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
- modification.merge(path, data);
- }
+ abstract DataTreeSnapshot readOnlyView();
- @Override
- void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
- modification.write(path, data);
- }
+ abstract void applyModifyTransactionRequest(ModifyTransactionRequest request,
+ @Nullable Consumer<Response<?, ?>> callback);
@Override
- CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path) {
- return Futures.immediateCheckedFuture(modification.readNode(path).isPresent());
+ final CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path) {
+ return Futures.immediateCheckedFuture(readOnlyView().readNode(path).isPresent());
}
@Override
- CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(final YangInstanceIdentifier path) {
- return Futures.immediateCheckedFuture(modification.readNode(path));
- }
-
- private RuntimeException abortedException() {
- return new IllegalStateException("Tracker " + identifier + " has been aborted");
- }
-
- private RuntimeException submittedException() {
- return new IllegalStateException("Tracker " + identifier + " has been submitted");
+ final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(final YangInstanceIdentifier path) {
+ return Futures.immediateCheckedFuture(readOnlyView().readNode(path));
}
@Override
- void doAbort() {
+ final void doAbort() {
sendAbort(new AbortLocalTransactionRequest(identifier, localActor()), response -> {
LOG.debug("Transaction {} abort completed with {}", identifier, response);
});
}
- @Override
- CommitLocalTransactionRequest commitRequest(final boolean coordinated) {
- final CommitLocalTransactionRequest ret = new CommitLocalTransactionRequest(identifier, localActor(),
- modification, coordinated);
- modification = new FailedDataTreeModification(this::submittedException);
- return ret;
- }
-
- @Override
- void doSeal() {
- modification.ready();
- sealedModification = modification;
- }
-
- DataTreeSnapshot getSnapshot() {
- Preconditions.checkState(sealedModification != null, "Proxy %s is not sealed yet", identifier);
- return sealedModification;
- }
-
- private void applyModifyTransactionRequest(final ModifyTransactionRequest request,
- final @Nullable Consumer<Response<?, ?>> callback) {
- for (TransactionModification mod : request.getModifications()) {
- if (mod instanceof TransactionWrite) {
- modification.write(mod.getPath(), ((TransactionWrite)mod).getData());
- } else if (mod instanceof TransactionMerge) {
- modification.merge(mod.getPath(), ((TransactionMerge)mod).getData());
- } else if (mod instanceof TransactionDelete) {
- modification.delete(mod.getPath());
- } else {
- throw new IllegalArgumentException("Unsupported modification " + mod);
- }
- }
-
- final java.util.Optional<PersistenceProtocol> maybeProtocol = request.getPersistenceProtocol();
- if (maybeProtocol.isPresent()) {
- seal();
- Verify.verify(callback != null, "Request {} has null callback", request);
-
- switch (maybeProtocol.get()) {
- case ABORT:
- sendAbort(callback);
- break;
- case SIMPLE:
- sendRequest(commitRequest(false), callback);
- break;
- case THREE_PHASE:
- sendRequest(commitRequest(true), callback);
- break;
- default:
- throw new IllegalArgumentException("Unhandled protocol " + maybeProtocol.get());
- }
- }
- }
-
@Override
void handleForwardedRemoteRequest(final TransactionRequest<?> request,
final @Nullable Consumer<Response<?, ?>> callback) {
- LOG.debug("Applying forwaded request {}", request);
-
if (request instanceof ModifyTransactionRequest) {
applyModifyTransactionRequest((ModifyTransactionRequest) request, callback);
+ } else if (request instanceof ReadTransactionRequest) {
+ final YangInstanceIdentifier path = ((ReadTransactionRequest) request).getPath();
+ final Optional<NormalizedNode<?, ?>> result = readOnlyView().readNode(path);
+ callback.accept(new ReadTransactionSuccess(request.getTarget(), request.getSequence(), result));
+ } else if (request instanceof ExistsTransactionRequest) {
+ final YangInstanceIdentifier path = ((ExistsTransactionRequest) request).getPath();
+ final boolean result = readOnlyView().readNode(path).isPresent();
+ callback.accept(new ExistsTransactionSuccess(request.getTarget(), request.getSequence(), result));
+ } else if (request instanceof TransactionPurgeRequest) {
+ purge();
} else {
throw new IllegalArgumentException("Unhandled request " + request);
}
@Override
void forwardToRemote(final RemoteProxyTransaction successor, final TransactionRequest<?> request,
- final Consumer<Response<?, ?>> callback) throws RequestException {
+ final Consumer<Response<?, ?>> callback) {
if (request instanceof CommitLocalTransactionRequest) {
final CommitLocalTransactionRequest req = (CommitLocalTransactionRequest) request;
final DataTreeModification mod = req.getModification();
}
});
- successor.seal();
+ successor.ensureSealed();
final ModifyTransactionRequest successorReq = successor.commitRequest(req.isCoordinated());
successor.sendRequest(successorReq, callback);
@Override
void forwardToLocal(final LocalProxyTransaction successor, final TransactionRequest<?> request,
- final Consumer<Response<?, ?>> callback) throws RequestException {
+ final Consumer<Response<?, ?>> callback) {
if (request instanceof AbortLocalTransactionRequest) {
successor.sendAbort(request, callback);
- } else if (request instanceof CommitLocalTransactionRequest) {
- successor.sendCommit((CommitLocalTransactionRequest)request, callback);
+ } else if (request instanceof TransactionPurgeRequest) {
+ successor.purge();
} else {
throw new IllegalArgumentException("Unhandled request" + request);
}
LOG.debug("Forwarded request {} to successor {}", request, successor);
}
- private void sendAbort(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
+ void sendAbort(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
sendRequest(request, callback);
- modification = new FailedDataTreeModification(this::abortedException);
- }
-
- private void sendCommit(final CommitLocalTransactionRequest request, final Consumer<Response<?, ?>> callback) {
- // Rebase old modification on new data tree.
- try (DataTreeModificationCursor cursor = modification.createCursor(YangInstanceIdentifier.EMPTY)) {
- request.getModification().applyToCursor(cursor);
- }
-
- seal();
- sendRequest(commitRequest(request.isCoordinated()), callback);
}
}
Code Review / controller.git / blobdiff