*
* @param <T> Message type
*/
-abstract class AbstractLocalTransactionRequest<T extends AbstractLocalTransactionRequest<T>>
+public abstract class AbstractLocalTransactionRequest<T extends AbstractLocalTransactionRequest<T>>
extends TransactionRequest<T> {
private static final long serialVersionUID = 1L;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.NotThreadSafe;
import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
+import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
for (Object obj : successfulRequests) {
if (obj instanceof TransactionRequest) {
LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
- successor.handleForwardedRemoteRequest((TransactionRequest<?>) obj, response -> { });
+ successor.replay((TransactionRequest<?>) obj, response -> { });
} else {
Verify.verify(obj instanceof IncrementSequence);
successor.incrementSequence(((IncrementSequence) obj).getDelta());
if (getIdentifier().equals(req.getTarget())) {
Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
LOG.debug("Forwarding queued request {} to successor {}", req, successor);
- successor.handleForwardedRemoteRequest((TransactionRequest<?>) req, e.getCallback());
+ successor.replay((TransactionRequest<?>) req, e.getCallback());
it.remove();
}
}
}
}
+ /**
+ * Invoked from {@link #replayMessages(AbstractProxyTransaction, Iterable)} to have successor adopt an in-flight
+ * request.
+ *
+ * <p>
+ * Note: this method is invoked by the predecessor on the successor.
+ *
+ * @param request Request which needs to be forwarded
+ * @param callback Callback to be invoked once the request completes
+ */
+ private void replay(TransactionRequest<?> request, Consumer<Response<?, ?>> callback) {
+ if (request instanceof AbstractLocalTransactionRequest) {
+ handleForwardedLocalRequest((AbstractLocalTransactionRequest<?>) request, callback);
+ } else {
+ handleForwardedRemoteRequest(request, callback);
+ }
+ }
+
// Called with the connection locked
final void finishReconnect() {
final SuccessorState local = getSuccessorState();
abstract TransactionRequest<?> commitRequest(boolean coordinated);
/**
- * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor. There is
- * no equivalent of this call from {@link LocalProxyTransaction} because it does not send a request until all
- * operations are packaged in the message.
+ * Replay a request originating in this proxy to a successor remote proxy.
+ */
+ abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
+ Consumer<Response<?, ?>> callback);
+
+ /**
+ * Replay a request originating in this proxy to a successor local proxy.
+ */
+ abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
+ Consumer<Response<?, ?>> callback);
+
+ /**
+ * Invoked from {@link LocalProxyTransaction} when it replays its successful requests to its successor.
*
* <p>
* Note: this method is invoked by the predecessor on the successor.
* @param request Request which needs to be forwarded
* @param callback Callback to be invoked once the request completes
*/
- abstract void handleForwardedRemoteRequest(TransactionRequest<?> request,
+ abstract void handleForwardedLocalRequest(AbstractLocalTransactionRequest<?> request,
@Nullable Consumer<Response<?, ?>> callback);
/**
- * Replay a request originating in this proxy to a successor remote proxy.
- */
- abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
- Consumer<Response<?, ?>> callback);
-
- /**
- * Replay a request originating in this proxy to a successor local proxy.
+ * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor.
+ *
+ * <p>
+ * Note: this method is invoked by the predecessor on the successor.
+ *
+ * @param request Request which needs to be forwarded
+ * @param callback Callback to be invoked once the request completes
*/
- abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
- Consumer<Response<?, ?>> callback);
+ abstract void handleForwardedRemoteRequest(TransactionRequest<?> request,
+ @Nullable Consumer<Response<?, ?>> callback);
@Override
public final String toString() {
import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.ExistsTransactionSuccess;
});
}
+ @Override
+ void handleForwardedLocalRequest(final AbstractLocalTransactionRequest<?> request,
+ final Consumer<Response<?, ?>> callback) {
+ if (request instanceof AbortLocalTransactionRequest) {
+ sendAbort(request, callback);
+ } else {
+ throw new IllegalArgumentException("Unhandled request" + request);
+ }
+ }
+
@Override
void handleForwardedRemoteRequest(final TransactionRequest<?> request,
final @Nullable Consumer<Response<?, ?>> callback) {
Verify.verify(protocol == PersistenceProtocol.ABORT);
abort();
}
-
}
import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.CommitLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
}
}
+ @Override
+ void handleForwardedLocalRequest(final AbstractLocalTransactionRequest<?> request,
+ final Consumer<Response<?, ?>> callback) {
+ if (request instanceof CommitLocalTransactionRequest) {
+ sendCommit((CommitLocalTransactionRequest) request, callback);
+ } else {
+ super.handleForwardedLocalRequest(request, callback);
+ }
+ }
+
@Override
void handleForwardedRemoteRequest(final TransactionRequest<?> request,
final @Nullable Consumer<Response<?, ?>> callback) {
@Override
void forwardRequest(final Request<?, ?> request, final Consumer<Response<?, ?>> callback,
final BiConsumer<Request<?, ?>, Consumer<Response<?, ?>>> forwardTo) throws RequestException {
+ // FIXME: do not use sendRequest() once we have throttling in place, as we have already waited the
+ // period required to get into the queue.
if (request instanceof TransactionRequest) {
forwardTransactionRequest((TransactionRequest<?>) request, callback);
} else if (request instanceof LocalHistoryRequest) {
import com.google.common.util.concurrent.SettableFuture;
import java.util.function.Consumer;
import javax.annotation.Nullable;
+import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.AbstractReadTransactionRequest;
+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.concepts.RequestFailure;
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.mdsal.common.api.ReadFailedException;
import org.opendaylight.yangtools.util.concurrent.MappingCheckedFuture;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
sendRequest(request, response -> completeModify(request, response));
}
+ @Override
+ void handleForwardedLocalRequest(final AbstractLocalTransactionRequest<?> request,
+ final Consumer<Response<?, ?>> callback) {
+ if (request instanceof CommitLocalTransactionRequest) {
+ replayLocalCommitRequest((CommitLocalTransactionRequest) request, callback);
+ } else if (request instanceof AbortLocalTransactionRequest) {
+ sendRequest(abortRequest(), callback);
+ } else {
+ throw new IllegalStateException("Unhandled request " + request);
+ }
+ }
+
+ private void replayLocalCommitRequest(final CommitLocalTransactionRequest request,
+ final Consumer<Response<?, ?>> callback) {
+ final DataTreeModification mod = request.getModification();
+ mod.applyToCursor(new AbstractDataTreeModificationCursor() {
+ @Override
+ public void write(final PathArgument child, final NormalizedNode<?, ?> data) {
+ doWrite(current().node(child), data);
+ }
+
+ @Override
+ public void merge(final PathArgument child, final NormalizedNode<?, ?> data) {
+ doMerge(current().node(child), data);
+ }
+
+ @Override
+ public void delete(final PathArgument child) {
+ doDelete(current().node(child));
+ }
+ });
+
+ sendRequest(commitRequest(request.isCoordinated()), callback);
+ }
+
@Override
void handleForwardedRemoteRequest(final TransactionRequest<?> request,
final @Nullable Consumer<Response<?, ?>> callback) {
recordFinishedRequest();
}
+ private ModifyTransactionRequest abortRequest() {
+ ensureInitializedBuilder();
+ builder.setAbort();
+ final ModifyTransactionRequest ret = builder.build();
+ builderBusy = false;
+ return ret;
+ }
+
@Override
ModifyTransactionRequest commitRequest(final boolean coordinated) {
ensureInitializedBuilder();
switch (maybeProto.get()) {
case ABORT:
- ensureInitializedBuilder();
- builder.setAbort();
- final ModifyTransactionRequest newReq = builder.build();
- builderBusy = false;
- sendRequest(newReq, callback);
+ sendRequest(abortRequest(), callback);
break;
case SIMPLE:
sendRequest(commitRequest(false), callback);
import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.ExistsTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
import org.opendaylight.controller.cluster.access.commands.ReadTransactionRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
transaction.recordSuccessfulRequest(successful2);
transaction.startReconnect();
transaction.replayMessages(successor.getTransaction(), entries);
- Assert.assertEquals(successful1, successor.expectTransactionRequest(AbortLocalTransactionRequest.class));
+
+ final ModifyTransactionRequest transformed = successor.expectTransactionRequest(ModifyTransactionRequest.class);
+ Assert.assertNotNull(transformed);
+ Assert.assertEquals(successful1.getSequence(), transformed.getSequence());
+ Assert.assertTrue(transformed.getPersistenceProtocol().isPresent());
+ Assert.assertEquals(PersistenceProtocol.ABORT, transformed.getPersistenceProtocol().get());
Assert.assertEquals(successful2, successor.expectTransactionRequest(ReadTransactionRequest.class));
Assert.assertEquals(request1, successor.expectTransactionRequest(ReadTransactionRequest.class));
Assert.assertEquals(request2, successor.expectTransactionRequest(ExistsTransactionRequest.class));
protected void checkModifications(final ModifyTransactionRequest modifyRequest) {
final List<TransactionModification> modifications = modifyRequest.getModifications();
Assert.assertEquals(3, modifications.size());
- Assert.assertThat(modifications, hasItem(both(isA(TransactionWrite.class)).and((hasPath(PATH_1)))));
- Assert.assertThat(modifications, hasItem(both(isA(TransactionMerge.class)).and((hasPath(PATH_2)))));
- Assert.assertThat(modifications, hasItem(both(isA(TransactionDelete.class)).and((hasPath(PATH_3)))));
+ Assert.assertThat(modifications, hasItem(both(isA(TransactionWrite.class)).and(hasPath(PATH_1))));
+ Assert.assertThat(modifications, hasItem(both(isA(TransactionMerge.class)).and(hasPath(PATH_2))));
+ Assert.assertThat(modifications, hasItem(both(isA(TransactionDelete.class)).and(hasPath(PATH_3))));
}
protected void testRequestResponse(final Consumer<VotingFuture> consumer,
@SuppressWarnings("unchecked")
protected <T> Consumer<T> createCallbackMock() {
- return (Consumer<T>) mock(Consumer.class);
+ return mock(Consumer.class);
}
protected static BaseMatcher<TransactionModification> hasPath(final YangInstanceIdentifier path) {
Code Review / controller.git / commitdiff