import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.Optional;
+import java.util.OptionalLong;
import java.util.function.Consumer;
-import javax.annotation.Nullable;
import org.opendaylight.controller.cluster.access.client.RequestTimeoutException;
import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
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.DataTreeModification;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@Override
void doDelete(final YangInstanceIdentifier path) {
- appendModification(new TransactionDelete(path), Optional.empty());
+ appendModification(new TransactionDelete(path), OptionalLong.empty());
}
@Override
- void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
- appendModification(new TransactionMerge(path, data), Optional.empty());
+ void doMerge(final YangInstanceIdentifier path, final NormalizedNode data) {
+ appendModification(new TransactionMerge(path, data), OptionalLong.empty());
}
@Override
- void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
- appendModification(new TransactionWrite(path, data), Optional.empty());
+ void doWrite(final YangInstanceIdentifier path, final NormalizedNode data) {
+ appendModification(new TransactionWrite(path, data), OptionalLong.empty());
}
private <T> FluentFuture<T> sendReadRequest(final AbstractReadTransactionRequest<?> request,
}
@Override
- FluentFuture<Optional<NormalizedNode<?, ?>>> doRead(final YangInstanceIdentifier path) {
- final SettableFuture<Optional<NormalizedNode<?, ?>>> future = SettableFuture.create();
+ FluentFuture<Optional<NormalizedNode>> doRead(final YangInstanceIdentifier path) {
+ final SettableFuture<Optional<NormalizedNode>> future = SettableFuture.create();
return sendReadRequest(new ReadTransactionRequest(getIdentifier(), nextSequence(), localActor(), path,
isSnapshotOnly()), t -> completeRead(path, future, t), future);
}
}
private void ensureFlushedBuider() {
- ensureFlushedBuider(Optional.empty());
+ ensureFlushedBuider(OptionalLong.empty());
}
- private void ensureFlushedBuider(final Optional<Long> enqueuedTicks) {
+ private void ensureFlushedBuider(final OptionalLong enqueuedTicks) {
if (builderBusy) {
flushBuilder(enqueuedTicks);
}
}
- private void flushBuilder(final Optional<Long> enqueuedTicks) {
+ private void flushBuilder(final OptionalLong enqueuedTicks) {
final ModifyTransactionRequest request = builder.build();
builderBusy = false;
sendModification(request, enqueuedTicks);
}
- private void sendModification(final TransactionRequest<?> request, final Optional<Long> enqueuedTicks) {
+ private void sendModification(final TransactionRequest<?> request, final OptionalLong enqueuedTicks) {
if (enqueuedTicks.isPresent()) {
- enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.get().longValue());
+ enqueueRequest(request, response -> completeModify(request, response), enqueuedTicks.getAsLong());
} else {
sendRequest(request, response -> completeModify(request, response));
}
}
private void appendModification(final TransactionModification modification) {
- appendModification(modification, Optional.empty());
+ appendModification(modification, OptionalLong.empty());
}
- private void appendModification(final TransactionModification modification, final Optional<Long> enqueuedTicks) {
+ private void appendModification(final TransactionModification modification, final OptionalLong enqueuedTicks) {
if (operationFailure == null) {
ensureInitializedBuilder();
recordFinishedRequest(response);
}
- private void completeRead(final YangInstanceIdentifier path,
- final SettableFuture<Optional<NormalizedNode<?, ?>>> future, final Response<?, ?> response) {
+ private void completeRead(final YangInstanceIdentifier path, final SettableFuture<Optional<NormalizedNode>> future,
+ final Response<?, ?> response) {
LOG.debug("Read request for {} completed with {}", path, response);
if (response instanceof ReadTransactionSuccess) {
}
@Override
- boolean sealAndSend(final Optional<Long> enqueuedTicks) {
+ boolean sealAndSend(final OptionalLong enqueuedTicks) {
if (sendReadyOnSeal) {
ensureInitializedBuilder();
builder.setReady();
} else if (request instanceof TransactionPurgeRequest) {
enqueuePurge(callback);
} else {
- throw new IllegalArgumentException("Unhandled request {}" + request);
+ throw unhandledRequest(request);
}
}
} else if (request instanceof AbortLocalTransactionRequest) {
enqueueRequest(abortRequest(), callback, enqueuedTicks);
} else {
- throw new IllegalStateException("Unhandled request " + request);
+ throw unhandledRequest(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));
+ final OptionalLong optTicks = OptionalLong.of(enqueuedTicks);
mod.applyToCursor(new AbstractDataTreeModificationCursor() {
@Override
- public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
+ 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) {
+ public void merge(final PathArgument child, final NormalizedNode data) {
appendModification(new TransactionMerge(current().node(child), data), optTicks);
}
}
@Override
- void handleReplayedRemoteRequest(final TransactionRequest<?> request,
- @Nullable final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+ void handleReplayedRemoteRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+ final long enqueuedTicks) {
final Consumer<Response<?, ?>> cb = callback != null ? callback : resp -> { /* NOOP */ };
- final Optional<Long> optTicks = Optional.of(Long.valueOf(enqueuedTicks));
+ final OptionalLong optTicks = OptionalLong.of(enqueuedTicks);
if (request instanceof ModifyTransactionRequest) {
handleReplayedModifyTransactionRequest(enqueuedTicks, cb, (ModifyTransactionRequest) request);
snapshotOnly, req.getIncrement()), callback, enqueuedTicks);
incrementSequence(req.getIncrement());
} else {
- throw new IllegalArgumentException("Unhandled request {}" + request);
+ throw unhandledRequest(request);
}
}