import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.BiConsumer;
import java.util.function.Consumer;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.access.client.AbstractClientConnection;
@Override
void onTransactionCompleted(final AbstractProxyTransaction tx) {
Verify.verify(tx instanceof LocalProxyTransaction);
- if (tx instanceof LocalReadWriteProxyTransaction) {
- if (LAST_SEALED_UPDATER.compareAndSet(this, (LocalReadWriteProxyTransaction) tx, null)) {
- LOG.debug("Completed last sealed transaction {}", tx);
- }
+ if (tx instanceof LocalReadWriteProxyTransaction
+ && LAST_SEALED_UPDATER.compareAndSet(this, (LocalReadWriteProxyTransaction) tx, null)) {
+ LOG.debug("Completed last sealed transaction {}", tx);
}
}
}
@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.
+ void replayEntry(final ConnectionEntry entry, final Consumer<ConnectionEntry> replayTo)
+ throws RequestException {
+ final Request<?, ?> request = entry.getRequest();
if (request instanceof TransactionRequest) {
- forwardTransactionRequest((TransactionRequest<?>) request, callback);
+ lookupProxy(request).replayRequest((TransactionRequest<?>) request, entry.getCallback(),
+ entry.getEnqueuedTicks());
} else if (request instanceof LocalHistoryRequest) {
- forwardTo.accept(request, callback);
+ replayTo.accept(entry);
} else {
throw new IllegalArgumentException("Unhandled request " + request);
}
}
- private void forwardTransactionRequest(final TransactionRequest<?> request,
- final Consumer<Response<?, ?>> callback) throws RequestException {
+ @Override
+ void forwardEntry(final ConnectionEntry entry, final Consumer<ConnectionEntry> forwardTo)
+ throws RequestException {
+ final Request<?, ?> request = entry.getRequest();
+ if (request instanceof TransactionRequest) {
+ lookupProxy(request).forwardRequest((TransactionRequest<?>) request, entry.getCallback());
+ } else if (request instanceof LocalHistoryRequest) {
+ forwardTo.accept(entry);
+ } else {
+ throw new IllegalArgumentException("Unhandled request " + request);
+ }
+ }
+ private AbstractProxyTransaction lookupProxy(final Request<?, ?> request)
+ throws RequestReplayException {
final AbstractProxyTransaction proxy;
lock.lock();
try {
} finally {
lock.unlock();
}
- if (proxy == null) {
- throw new RequestReplayException("Failed to find proxy for %s", request);
+ if (proxy != null) {
+ return proxy;
}
- proxy.forwardRequest(request, callback);
+ throw new RequestReplayException("Failed to find proxy for %s", request);
}
}
}
@GuardedBy("lock")
+ @SuppressWarnings("checkstyle:hiddenField")
abstract AbstractProxyTransaction doCreateTransactionProxy(AbstractClientConnection<ShardBackendInfo> connection,
TransactionIdentifier txId, boolean snapshotOnly, boolean isDone);
+ @SuppressWarnings("checkstyle:hiddenField")
abstract ProxyHistory createSuccessor(AbstractClientConnection<ShardBackendInfo> connection);
@SuppressFBWarnings(value = "UL_UNRELEASED_LOCK", justification = "Lock is released asynchronously via the cohort")