*/
package org.opendaylight.controller.cluster.datastore;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
import com.google.common.primitives.UnsignedLong;
import com.google.common.util.concurrent.FutureCallback;
import java.util.Collection;
-import javax.annotation.Nullable;
-import javax.annotation.concurrent.NotThreadSafe;
+import java.util.Optional;
+import org.eclipse.jdt.annotation.Nullable;
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.concepts.RuntimeRequestException;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.access.concepts.UnsupportedRequestException;
+import org.opendaylight.yangtools.yang.common.Empty;
import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Frontend read-write transaction state as observed by the shard leader.
+ * Frontend read-write transaction state as observed by the shard leader. This class is NOT thread-safe.
*
* @author Robert Varga
*/
-@NotThreadSafe
final class FrontendReadWriteTransaction extends FrontendTransaction {
private enum CommitStage {
READY,
final RequestException cause;
Failed(final RequestException cause) {
- this.cause = Preconditions.checkNotNull(cause);
+ this.cause = requireNonNull(cause);
}
@Override
final ReadWriteShardDataTreeTransaction openTransaction;
Open(final ReadWriteShardDataTreeTransaction openTransaction) {
- this.openTransaction = Preconditions.checkNotNull(openTransaction);
+ this.openTransaction = requireNonNull(openTransaction);
}
@Override
CommitStage stage;
Ready(final ShardDataTreeCohort readyCohort) {
- this.readyCohort = Preconditions.checkNotNull(readyCohort);
- this.stage = CommitStage.READY;
+ this.readyCohort = requireNonNull(readyCohort);
+ stage = CommitStage.READY;
}
@Override
final DataTreeModification sealedModification;
Sealed(final DataTreeModification sealedModification) {
- this.sealedModification = Preconditions.checkNotNull(sealedModification);
+ this.sealedModification = requireNonNull(sealedModification);
}
@Override
private FrontendReadWriteTransaction(final AbstractFrontendHistory history, final TransactionIdentifier id,
final ReadWriteShardDataTreeTransaction transaction) {
super(history, id);
- this.state = new Open(transaction);
+ state = new Open(transaction);
}
private FrontendReadWriteTransaction(final AbstractFrontendHistory history, final TransactionIdentifier id,
final DataTreeModification mod) {
super(history, id);
- this.state = new Sealed(mod);
+ state = new Sealed(mod);
}
static FrontendReadWriteTransaction createOpen(final AbstractFrontendHistory history,
// Sequence has already been checked
@Override
- @Nullable TransactionSuccess<?> doHandleRequest(final TransactionRequest<?> request, final RequestEnvelope envelope,
+ TransactionSuccess<?> doHandleRequest(final TransactionRequest<?> request, final RequestEnvelope envelope,
final long now) throws RequestException {
- if (request instanceof ModifyTransactionRequest) {
- return handleModifyTransaction((ModifyTransactionRequest) request, envelope, now);
- } else if (request instanceof CommitLocalTransactionRequest) {
- handleCommitLocalTransaction((CommitLocalTransactionRequest) request, envelope, now);
+ if (request instanceof ModifyTransactionRequest modifyRequest) {
+ return handleModifyTransaction(modifyRequest, envelope, now);
+ } else if (request instanceof CommitLocalTransactionRequest commitLocalRequest) {
+ handleCommitLocalTransaction(commitLocalRequest, envelope, now);
return null;
- } else if (request instanceof ExistsTransactionRequest) {
- return handleExistsTransaction((ExistsTransactionRequest) request);
- } else if (request instanceof ReadTransactionRequest) {
- return handleReadTransaction((ReadTransactionRequest) request);
- } else if (request instanceof TransactionPreCommitRequest) {
- handleTransactionPreCommit((TransactionPreCommitRequest) request, envelope, now);
+ } else if (request instanceof ExistsTransactionRequest existsRequest) {
+ return handleExistsTransaction(existsRequest);
+ } else if (request instanceof ReadTransactionRequest readRequest) {
+ return handleReadTransaction(readRequest);
+ } else if (request instanceof TransactionPreCommitRequest preCommitRequest) {
+ handleTransactionPreCommit(preCommitRequest, envelope, now);
return null;
- } else if (request instanceof TransactionDoCommitRequest) {
- handleTransactionDoCommit((TransactionDoCommitRequest) request, envelope, now);
+ } else if (request instanceof TransactionDoCommitRequest doCommitRequest) {
+ handleTransactionDoCommit(doCommitRequest, envelope, now);
return null;
} else if (request instanceof TransactionAbortRequest) {
return handleTransactionAbort(request.getSequence(), envelope, now);
final Ready ready = checkReady();
startAbort();
- ready.readyCohort.abort(new FutureCallback<Void>() {
+ ready.readyCohort.abort(new FutureCallback<>() {
@Override
- public void onSuccess(final Void result) {
+ public void onSuccess(final Empty result) {
recordAndSendSuccess(envelope, now, new TransactionAbortSuccess(getIdentifier(), sequence));
finishAbort();
}
case READY:
ready.stage = CommitStage.CAN_COMMIT_PENDING;
LOG.debug("{}: Transaction {} initiating canCommit", persistenceId(), getIdentifier());
- checkReady().readyCohort.canCommit(new FutureCallback<Void>() {
+ checkReady().readyCohort.canCommit(new FutureCallback<>() {
@Override
- public void onSuccess(final Void result) {
+ public void onSuccess(final Empty result) {
successfulCanCommit(envelope, now);
}
case READY:
ready.stage = CommitStage.CAN_COMMIT_PENDING;
LOG.debug("{}: Transaction {} initiating direct canCommit", persistenceId(), getIdentifier());
- ready.readyCohort.canCommit(new FutureCallback<Void>() {
+ ready.readyCohort.canCommit(new FutureCallback<>() {
@Override
- public void onSuccess(final Void result) {
+ public void onSuccess(final Empty result) {
successfulDirectCanCommit(envelope, now);
}
throw new UnsupportedRequestException(request);
}
- final java.util.Optional<Exception> optFailure = request.getDelayedFailure();
+ final Optional<Exception> optFailure = request.getDelayedFailure();
if (optFailure.isPresent()) {
- state = new Ready(history().createFailedCohort(getIdentifier(), sealedModification, optFailure.get()));
+ state = new Ready(history().createFailedCohort(getIdentifier(), sealedModification,
+ optFailure.orElseThrow()));
} else {
- state = new Ready(history().createReadyCohort(getIdentifier(), sealedModification));
+ state = new Ready(history().createReadyCohort(getIdentifier(), sealedModification, Optional.empty()));
}
if (request.isCoordinated()) {
}
}
- private ExistsTransactionSuccess handleExistsTransaction(final ExistsTransactionRequest request)
- throws RequestException {
- final Optional<NormalizedNode<?, ?>> data = checkOpen().getSnapshot().readNode(request.getPath());
+ private ExistsTransactionSuccess handleExistsTransaction(final ExistsTransactionRequest request) {
+ final Optional<NormalizedNode> data = checkOpen().getSnapshot().readNode(request.getPath());
return recordSuccess(request.getSequence(), new ExistsTransactionSuccess(getIdentifier(), request.getSequence(),
data.isPresent()));
}
- private ReadTransactionSuccess handleReadTransaction(final ReadTransactionRequest request)
- throws RequestException {
- final Optional<NormalizedNode<?, ?>> data = checkOpen().getSnapshot().readNode(request.getPath());
+ private ReadTransactionSuccess handleReadTransaction(final ReadTransactionRequest request) {
+ final Optional<NormalizedNode> data = checkOpen().getSnapshot().readNode(request.getPath());
return recordSuccess(request.getSequence(), new ReadTransactionSuccess(getIdentifier(), request.getSequence(),
data));
}
for (TransactionModification m : modifications) {
if (m instanceof TransactionDelete) {
modification.delete(m.getPath());
- } else if (m instanceof TransactionWrite) {
- modification.write(m.getPath(), ((TransactionWrite) m).getData());
- } else if (m instanceof TransactionMerge) {
- modification.merge(m.getPath(), ((TransactionMerge) m).getData());
+ } else if (m instanceof TransactionWrite write) {
+ modification.write(m.getPath(), write.getData());
+ } else if (m instanceof TransactionMerge merge) {
+ modification.merge(m.getPath(), merge.getData());
} else {
LOG.warn("{}: ignoring unhandled modification {}", persistenceId(), m);
}
}
}
- @Nullable
- private TransactionSuccess<?> handleModifyTransaction(final ModifyTransactionRequest request,
+ private @Nullable TransactionSuccess<?> handleModifyTransaction(final ModifyTransactionRequest request,
final RequestEnvelope envelope, final long now) throws RequestException {
// We need to examine the persistence protocol first to see if this is an idempotent request. If there is no
// protocol, there is nothing for us to do.
- final
java.util.Optional<PersistenceProtocol> maybeProto = request.getPersistenceProtocol();
+ final Optional<PersistenceProtocol> maybeProto = request.getPersistenceProtocol();
if (!maybeProto.isPresent()) {
applyModifications(request.getModifications());
return replyModifySuccess(request.getSequence());
}
- switch (maybeProto.get()) {
+ switch (maybeProto.orElseThrow()) {
case ABORT:
if (ABORTING.equals(state)) {
LOG.debug("{}: Transaction {} already aborting", persistenceId(), getIdentifier());
coordinatedCommit(envelope, now);
return null;
default:
- LOG.warn("{}: rejecting unsupported protocol {}", persistenceId(), maybeProto.get());
+ LOG.warn("{}: rejecting unsupported protocol {}", persistenceId(), maybeProto.orElseThrow());
throw new UnsupportedRequestException(request);
}
}
}
applyModifications(modifications);
- state = new Ready(checkOpen().ready());
+ state = new Ready(checkOpen().ready(Optional.empty()));
LOG.debug("{}: transitioned {} to ready", persistenceId(), getIdentifier());
}
}
private ReadWriteShardDataTreeTransaction checkOpen() {
- Preconditions.checkState(state instanceof Open, "%s expect to be open, is in state %s", getIdentifier(),
- state);
+ checkState(state instanceof Open, "%s expect to be open, is in state %s", getIdentifier(), state);
return ((Open) state).openTransaction;
}
private Ready checkReady() {
- Preconditions.checkState(state instanceof Ready, "%s expect to be ready, is in state %s", getIdentifier(),
- state);
+ checkState(state instanceof Ready, "%s expect to be ready, is in state %s", getIdentifier(), state);
return (Ready) state;
}
private DataTreeModification checkSealed() {
- Preconditions.checkState(state instanceof Sealed, "%s expect to be sealed, is in state %s", getIdentifier(),
- state);
+ checkState(state instanceof Sealed, "%s expect to be sealed, is in state %s", getIdentifier(), state);
return ((Sealed) state).sealedModification;
}
Code Review / controller.git / blobdiff