import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionSuccess;
import org.opendaylight.controller.cluster.access.commands.TransactionWrite;
-import org.opendaylight.controller.cluster.access.concepts.RequestException;
import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
import org.opendaylight.controller.cluster.access.concepts.Response;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
private static final int REQUEST_MAX_MODIFICATIONS = 1000;
private final ModifyTransactionRequestBuilder builder;
+ private final boolean sendReadyOnSeal;
+ private final boolean snapshotOnly;
private boolean builderBusy;
private volatile Exception operationFailure;
- RemoteProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier) {
+
+ RemoteProxyTransaction(final ProxyHistory parent, final TransactionIdentifier identifier,
+ final boolean snapshotOnly, final boolean sendReadyOnSeal) {
super(parent);
+ this.snapshotOnly = snapshotOnly;
+ this.sendReadyOnSeal = sendReadyOnSeal;
builder = new ModifyTransactionRequestBuilder(identifier, localActor());
}
+ @Override
+ boolean isSnapshotOnly() {
+ return snapshotOnly;
+ }
+
@Override
public TransactionIdentifier getIdentifier() {
return builder.getIdentifier();
@Override
CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path) {
final SettableFuture<Boolean> future = SettableFuture.create();
- return sendReadRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(), path),
- t -> completeExists(future, t), future);
+ return sendReadRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(), path,
+ isSnapshotOnly()), t -> completeExists(future, t), future);
}
@Override
CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(final YangInstanceIdentifier path) {
final SettableFuture<Optional<NormalizedNode<?, ?>>> future = SettableFuture.create();
- return sendReadRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), path),
- t -> completeRead(future, t), future);
+ return sendReadRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), path,
+ isSnapshotOnly()), t -> completeRead(future, t), future);
}
@Override
void doAbort() {
- ensureInitializedBuider();
+ ensureInitializedBuilder();
builder.setAbort();
flushBuilder();
}
- private void ensureInitializedBuider() {
+ private void ensureInitializedBuilder() {
if (!builderBusy) {
builder.setSequence(nextSequence());
builderBusy = true;
private void appendModification(final TransactionModification modification) {
if (operationFailure == null) {
- ensureInitializedBuider();
+ ensureInitializedBuilder();
builder.addModification(modification);
if (builder.size() >= REQUEST_MAX_MODIFICATIONS) {
@Override
ModifyTransactionRequest commitRequest(final boolean coordinated) {
- ensureInitializedBuider();
+ ensureInitializedBuilder();
builder.setCommit(coordinated);
final ModifyTransactionRequest ret = builder.build();
@Override
void doSeal() {
- // No-op
+ if (sendReadyOnSeal) {
+ ensureInitializedBuilder();
+ builder.setReady();
+ flushBuilder();
+ }
+ }
+
+ @Override
+ void flushState(final AbstractProxyTransaction successor) {
+ if (builderBusy) {
+ final ModifyTransactionRequest request = builder.build();
+ builderBusy = false;
+ successor.handleForwardedRemoteRequest(request, null);
+ }
}
@Override
void forwardToRemote(final RemoteProxyTransaction successor, final TransactionRequest<?> request,
- final Consumer<Response<?, ?>> callback) throws RequestException {
+ final Consumer<Response<?, ?>> callback) {
successor.handleForwardedRequest(request, callback);
}
- private void handleForwardedRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback)
- throws RequestException {
+ private void handleForwardedRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
if (request instanceof ModifyTransactionRequest) {
final ModifyTransactionRequest req = (ModifyTransactionRequest) request;
final java.util.Optional<PersistenceProtocol> maybeProto = req.getPersistenceProtocol();
if (maybeProto.isPresent()) {
- seal();
+ ensureSealed();
switch (maybeProto.get()) {
case ABORT:
} else if (request instanceof ReadTransactionRequest) {
ensureFlushedBuider();
sendRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(),
- ((ReadTransactionRequest) request).getPath()), callback);
+ ((ReadTransactionRequest) request).getPath(), isSnapshotOnly()), callback);
} else if (request instanceof ExistsTransactionRequest) {
ensureFlushedBuider();
sendRequest(new ExistsTransactionRequest(getIdentifier(), nextSequence(), localActor(),
- ((ExistsTransactionRequest) request).getPath()), callback);
+ ((ExistsTransactionRequest) request).getPath(), isSnapshotOnly()), callback);
} else if (request instanceof TransactionPreCommitRequest) {
ensureFlushedBuider();
sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), callback);
@Override
void forwardToLocal(final LocalProxyTransaction successor, final TransactionRequest<?> request,
- final Consumer<Response<?, ?>> callback) throws RequestException {
+ final Consumer<Response<?, ?>> callback) {
successor.handleForwardedRemoteRequest(request, callback);
}
}