X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatabroker%2Factors%2Fdds%2FClientLocalHistory.java;h=493eb4089eb86cf9f119bfdf5589f780404b3715;hb=abaef4a5ae37f27542155457fe7306a4662b1eeb;hp=807cf98cb73b3dcf5e4841eaf0f4d5b171ac025f;hpb=a510fba141230ce9fe8301f9eb0198cc09df46ca;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientLocalHistory.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientLocalHistory.java index 807cf98cb7..493eb4089e 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientLocalHistory.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/ClientLocalHistory.java @@ -9,8 +9,7 @@ package org.opendaylight.controller.cluster.databroker.actors.dds; import com.google.common.annotations.Beta; import com.google.common.base.Preconditions; -import com.google.common.base.Verify; -import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import org.opendaylight.controller.cluster.access.client.AbstractClientConnection; import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier; import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier; @@ -20,48 +19,75 @@ import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier * *

* This interface is used by the world outside of the actor system and in the actor system it is manifested via - * its client actor. That requires some state transfer with {@link DistributedDataStoreClientBehavior}. In order to + * its client actor. That requires some state transfer with {@link AbstractDataStoreClientBehavior}. In order to * reduce request latency, all messages are carbon-copied (and enqueued first) to the client actor. * * @author Robert Varga */ @Beta -public final class ClientLocalHistory extends AbstractClientHistory implements AutoCloseable { - - private static final AtomicLongFieldUpdater NEXT_TX_UPDATER = - AtomicLongFieldUpdater.newUpdater(ClientLocalHistory.class, "nextTx"); - - // Used via NEXT_TX_UPDATER - @SuppressWarnings("unused") - private volatile long nextTx = 0; - - ClientLocalHistory(final DistributedDataStoreClientBehavior client, final LocalHistoryIdentifier historyId) { +public class ClientLocalHistory extends AbstractClientHistory implements AutoCloseable { + ClientLocalHistory(final AbstractDataStoreClientBehavior client, final LocalHistoryIdentifier historyId) { super(client, historyId); } - public ClientTransaction createTransaction() { + @Override + public void close() { + doClose(); + } + + private State ensureIdleState() { final State local = state(); Preconditions.checkState(local == State.IDLE, "Local history %s state is %s", this, local); - updateState(local, State.TX_OPEN); + return local; + } - return new ClientTransaction(this, new TransactionIdentifier(getIdentifier(), - NEXT_TX_UPDATER.getAndIncrement(this))); + @Override + ClientSnapshot doCreateSnapshot() { + ensureIdleState(); + return new ClientSnapshot(this, new TransactionIdentifier(getIdentifier(), nextTx())); } @Override - public void close() { - final State local = state(); - if (local != State.CLOSED) { - Preconditions.checkState(local == State.IDLE, "Local history %s has an open transaction", this); - updateState(local, State.CLOSED); + ClientTransaction doCreateTransaction() { + updateState(ensureIdleState(), State.TX_OPEN); + return new ClientTransaction(this, new TransactionIdentifier(getIdentifier(), nextTx())); + } + + @Override + void onTransactionAbort(final AbstractClientHandle snap) { + if (snap instanceof ClientTransaction) { + final State local = state(); + if (local == State.TX_OPEN) { + updateState(local, State.IDLE); + } } + + super.onTransactionAbort(snap); } @Override - void onTransactionReady(final ClientTransaction transaction) { + AbstractTransactionCommitCohort onTransactionReady(final ClientTransaction tx, + final AbstractTransactionCommitCohort cohort) { + final State local = state(); - Verify.verify(local == State.TX_OPEN, "Local history %s is in unexpected state %s", this, local); - updateState(local, State.IDLE); - super.onTransactionReady(transaction); + switch (local) { + case CLOSED: + return super.onTransactionReady(tx, cohort); + case IDLE: + throw new IllegalStateException(String.format("Local history %s is idle when readying transaction %s", + this, tx.getIdentifier())); + case TX_OPEN: + updateState(local, State.IDLE); + return super.onTransactionReady(tx, cohort); + default: + throw new IllegalStateException(String.format("Local history %s in unhandled state %s", this, local)); + + } + } + + @Override + ProxyHistory createHistoryProxy(final LocalHistoryIdentifier historyId, + final AbstractClientConnection connection) { + return ProxyHistory.createClient(this, connection, historyId); } }