ActorContext is overloaded name even within Akka, without us adding
to it. Furthermore Akka's AbstractActor defines ActorContext as its
nested class, which thoroughly breaks resolution priorities.
Rename ActorContext to ActorUtils, reducing confusion around class
uses.
Change-Id: I140239a8f74ee7deecf9ee848df0cfbbb72f3c4d
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>
68 files changed:
import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
this.serializer = serializer;
this.makeLeaderLocalTimeout =
this.serializer = serializer;
this.makeLeaderLocalTimeout =
- new Timeout(configDataStore.getActorContext().getDatastoreContext()
+ new Timeout(configDataStore.getActorUtils().getDatastoreContext()
.getShardLeaderElectionTimeout().duration().$times(2));
}
.getShardLeaderElectionTimeout().duration().$times(2));
}
return newFailedRpcResultFuture("A valid DataStoreType must be specified");
}
return newFailedRpcResultFuture("A valid DataStoreType must be specified");
}
- ActorContext actorContext = dataStoreType == DataStoreType.Config
- ? configDataStore.getActorContext()
- : operDataStore.getActorContext();
+ ActorUtils actorUtils = dataStoreType == DataStoreType.Config
+ ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
- actorContext.getCurrentMemberName().getName(), shardName, dataStoreType);
+ actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
final scala.concurrent.Future<ActorRef> localShardReply =
final scala.concurrent.Future<ActorRef> localShardReply =
- actorContext.findLocalShardAsync(shardName);
+ actorUtils.findLocalShardAsync(shardName);
final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
localShardReply.onComplete(new OnComplete<ActorRef>() {
final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
localShardReply.onComplete(new OnComplete<ActorRef>() {
makeLeaderLocalAsk.failure(failure);
} else {
makeLeaderLocalAsk
makeLeaderLocalAsk.failure(failure);
} else {
makeLeaderLocalAsk
- .completeWith(actorContext
+ .completeWith(actorUtils
.executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
}
}
.executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
LOG.debug("Leadership transfer complete");
future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
}
LOG.debug("Leadership transfer complete");
future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
final Function<String, Object> messageSupplier) {
private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
final Function<String, Object> messageSupplier) {
- ActorContext actorContext = dataStoreType == DataStoreType.Config ? configDataStore.getActorContext()
- : operDataStore.getActorContext();
- Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
+ ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
+ : operDataStore.getActorUtils();
+ Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
- LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
+ LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
for (String shardName: allShardNames) {
for (String shardName: allShardNames) {
- ListenableFuture<T> future = this.ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
+ ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
SHARD_MGR_TIMEOUT);
shardResultData.add(new SimpleEntry<>(future,
new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
SHARD_MGR_TIMEOUT);
shardResultData.add(new SimpleEntry<>(future,
new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
Timeout timeout = SHARD_MGR_TIMEOUT;
private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
Timeout timeout = SHARD_MGR_TIMEOUT;
- ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
- ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
+ ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
+ ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
return Futures.allAsList(configFuture, operFuture);
}
private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
ActorRef shardManager = dataStoreType == DataStoreType.Config
return Futures.allAsList(configFuture, operFuture);
}
private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
ActorRef shardManager = dataStoreType == DataStoreType.Config
- ? configDataStore.getActorContext().getShardManager()
- : operDataStore.getActorContext().getShardManager();
+ ? configDataStore.getActorUtils().getShardManager()
+ : operDataStore.getActorUtils().getShardManager();
return ask(shardManager, message, SHARD_MGR_TIMEOUT);
}
return ask(shardManager, message, SHARD_MGR_TIMEOUT);
}
returnFuture.set(resp);
}
}
returnFuture.set(resp);
}
}
- }, configDataStore.getActorContext().getClientDispatcher());
+ }, configDataStore.getActorUtils().getClientDispatcher());
ImmutableMap<String, DatastoreSnapshot> map = ImmutableMap.of(snapshots.get(0).getType(), snapshots.get(0),
snapshots.get(1).getType(), snapshots.get(1));
ImmutableMap<String, DatastoreSnapshot> map = ImmutableMap.of(snapshots.get(0).getType(), snapshots.get(0),
snapshots.get(1).getType(), snapshots.get(1));
- verifyDatastoreSnapshot(node.configDataStore().getActorContext().getDataStoreName(),
- map.get(node.configDataStore().getActorContext().getDataStoreName()), "cars", "people");
+ verifyDatastoreSnapshot(node.configDataStore().getActorUtils().getDataStoreName(),
+ map.get(node.configDataStore().getActorUtils().getDataStoreName()), "cars", "people");
} finally {
new File(fileName).delete();
}
// Test failure by killing a shard.
} finally {
new File(fileName).delete();
}
// Test failure by killing a shard.
- node.configDataStore().getActorContext().getShardManager().tell(node.datastoreContextBuilder()
+ node.configDataStore().getActorUtils().getShardManager().tell(node.datastoreContextBuilder()
.shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(), ActorRef.noSender());
.shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(), ActorRef.noSender());
- ActorRef carsShardActor = node.configDataStore().getActorContext().findLocalShard("cars").get();
+ ActorRef carsShardActor = node.configDataStore().getActorUtils().findLocalShard("cars").get();
node.kit().watch(carsShardActor);
carsShardActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
node.kit().expectTerminated(carsShardActor);
node.kit().watch(carsShardActor);
carsShardActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
node.kit().expectTerminated(carsShardActor);
replicaNode2.kit().waitForMembersUp("member-1", "member-3");
replicaNode3.kit().waitForMembersUp("member-1", "member-2");
replicaNode2.kit().waitForMembersUp("member-1", "member-3");
replicaNode3.kit().waitForMembersUp("member-1", "member-2");
- final ActorRef shardManager1 = member1.configDataStore().getActorContext().getShardManager();
+ final ActorRef shardManager1 = member1.configDataStore().getActorUtils().getShardManager();
shardManager1.tell(new PrefixShardCreated(new PrefixShardConfiguration(
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH),
"prefix", Collections.singleton(MEMBER_1))),
ActorRef.noSender());
shardManager1.tell(new PrefixShardCreated(new PrefixShardConfiguration(
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH),
"prefix", Collections.singleton(MEMBER_1))),
ActorRef.noSender());
- member1.kit().waitUntilLeader(member1.configDataStore().getActorContext(),
+ member1.kit().waitUntilLeader(member1.configDataStore().getActorUtils(),
ClusterUtils.getCleanShardName(CarsModel.BASE_PATH));
final InstanceIdentifier<Cars> identifier = InstanceIdentifier.create(Cars.class);
ClusterUtils.getCleanShardName(CarsModel.BASE_PATH));
final InstanceIdentifier<Cars> identifier = InstanceIdentifier.create(Cars.class);
final MemberNode member1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name)
.moduleShardsConfig(moduleShardsConfig).build();
final MemberNode member1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name)
.moduleShardsConfig(moduleShardsConfig).build();
- member1.kit().waitUntilLeader(member1.configDataStore().getActorContext(), "default");
+ member1.kit().waitUntilLeader(member1.configDataStore().getActorUtils(), "default");
final RpcResult<GetShardRoleOutput> successResult =
getShardRole(member1, Mockito.mock(BindingNormalizedNodeSerializer.class), "default");
final RpcResult<GetShardRoleOutput> successResult =
getShardRole(member1, Mockito.mock(BindingNormalizedNodeSerializer.class), "default");
verifyFailedRpcResult(failedResult);
verifyFailedRpcResult(failedResult);
- final ActorRef shardManager1 = member1.configDataStore().getActorContext().getShardManager();
+ final ActorRef shardManager1 = member1.configDataStore().getActorUtils().getShardManager();
shardManager1.tell(new PrefixShardCreated(new PrefixShardConfiguration(
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH),
"prefix", Collections.singleton(MEMBER_1))),
ActorRef.noSender());
shardManager1.tell(new PrefixShardCreated(new PrefixShardConfiguration(
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH),
"prefix", Collections.singleton(MEMBER_1))),
ActorRef.noSender());
- member1.kit().waitUntilLeader(member1.configDataStore().getActorContext(),
+ member1.kit().waitUntilLeader(member1.configDataStore().getActorUtils(),
ClusterUtils.getCleanShardName(CarsModel.BASE_PATH));
final InstanceIdentifier<Cars> identifier = InstanceIdentifier.create(Cars.class);
ClusterUtils.getCleanShardName(CarsModel.BASE_PATH));
final InstanceIdentifier<Cars> identifier = InstanceIdentifier.create(Cars.class);
final MemberNode member1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name)
.moduleShardsConfig(moduleShardsConfig).build();
final MemberNode member1 = MemberNode.builder(memberNodes).akkaConfig("Member1").testName(name)
.moduleShardsConfig(moduleShardsConfig).build();
- member1.kit().waitUntilLeader(member1.configDataStore().getActorContext(), "default");
+ member1.kit().waitUntilLeader(member1.configDataStore().getActorUtils(), "default");
verifySuccessfulRpcResult(rpcResult);
verifyRaftPeersPresent(memberNode.configDataStore(), shardName, peerMemberNames);
verifySuccessfulRpcResult(rpcResult);
verifyRaftPeersPresent(memberNode.configDataStore(), shardName, peerMemberNames);
- Optional<ActorRef> optional = memberNode.configDataStore().getActorContext().findLocalShard(shardName);
+ Optional<ActorRef> optional = memberNode.configDataStore().getActorUtils().findLocalShard(shardName);
assertTrue("Replica shard not present", optional.isPresent());
}
assertTrue("Replica shard not present", optional.isPresent());
}
verifyRaftPeersPresent(memberNode.configDataStore(), shardName, peerMemberNames);
verifyRaftPeersPresent(memberNode.configDataStore(), shardName, peerMemberNames);
- Optional<ActorRef> optional = memberNode.operDataStore().getActorContext().findLocalShard(shardName);
+ Optional<ActorRef> optional = memberNode.operDataStore().getActorUtils().findLocalShard(shardName);
assertFalse("Oper shard present", optional.isPresent());
rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName(shardName)
assertFalse("Oper shard present", optional.isPresent());
rpcResult = service.addShardReplica(new AddShardReplicaInputBuilder().setShardName(shardName)
ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module",
"pets", null,
Collections.singletonList(MEMBER_1));
ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module",
"pets", null,
Collections.singletonList(MEMBER_1));
- leaderNode1.configDataStore().getActorContext().getShardManager().tell(
+ leaderNode1.configDataStore().getActorUtils().getShardManager().tell(
new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef());
leaderNode1.kit().expectMsgClass(Success.class);
new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef());
leaderNode1.kit().expectMsgClass(Success.class);
- leaderNode1.kit().waitUntilLeader(leaderNode1.configDataStore().getActorContext(), "pets");
+ leaderNode1.kit().waitUntilLeader(leaderNode1.configDataStore().getActorUtils(), "pets");
MemberNode newReplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name)
.moduleShardsConfig(moduleShardsConfig).build();
MemberNode newReplicaNode2 = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name)
.moduleShardsConfig(moduleShardsConfig).build();
leaderNode1.waitForMembersUp("member-2");
newReplicaNode2.waitForMembersUp("member-1");
leaderNode1.waitForMembersUp("member-2");
newReplicaNode2.waitForMembersUp("member-1");
- newReplicaNode2.configDataStore().getActorContext().getShardManager().tell(
+ newReplicaNode2.configDataStore().getActorUtils().getShardManager().tell(
new CreateShard(petsModuleConfig, Shard.builder(), null), newReplicaNode2.kit().getRef());
newReplicaNode2.kit().expectMsgClass(Success.class);
new CreateShard(petsModuleConfig, Shard.builder(), null), newReplicaNode2.kit().getRef());
newReplicaNode2.kit().expectMsgClass(Success.class);
- newReplicaNode2.operDataStore().getActorContext().getShardManager().tell(
+ newReplicaNode2.operDataStore().getActorUtils().getShardManager().tell(
new CreateShard(new ModuleShardConfiguration(URI.create("no-leader-ns"), "no-leader-module",
"no-leader", null,
Collections.singletonList(MEMBER_1)),
new CreateShard(new ModuleShardConfiguration(URI.create("no-leader-ns"), "no-leader-module",
"no-leader", null,
Collections.singletonList(MEMBER_1)),
ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module",
"pets", null, Arrays.asList(MEMBER_1, MEMBER_2, MEMBER_3));
ModuleShardConfiguration petsModuleConfig = new ModuleShardConfiguration(URI.create("pets-ns"), "pets-module",
"pets", null, Arrays.asList(MEMBER_1, MEMBER_2, MEMBER_3));
- leaderNode1.configDataStore().getActorContext().getShardManager().tell(
+ leaderNode1.configDataStore().getActorUtils().getShardManager().tell(
new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef());
leaderNode1.kit().expectMsgClass(Success.class);
new CreateShard(petsModuleConfig, Shard.builder(), null), leaderNode1.kit().getRef());
leaderNode1.kit().expectMsgClass(Success.class);
- replicaNode2.configDataStore().getActorContext().getShardManager().tell(
+ replicaNode2.configDataStore().getActorUtils().getShardManager().tell(
new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode2.kit().getRef());
replicaNode2.kit().expectMsgClass(Success.class);
new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode2.kit().getRef());
replicaNode2.kit().expectMsgClass(Success.class);
- replicaNode3.configDataStore().getActorContext().getShardManager().tell(
+ replicaNode3.configDataStore().getActorUtils().getShardManager().tell(
new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode3.kit().getRef());
replicaNode3.kit().expectMsgClass(Success.class);
new CreateShard(petsModuleConfig, Shard.builder(), null), replicaNode3.kit().getRef());
replicaNode3.kit().expectMsgClass(Success.class);
@SafeVarargs
private static void verifyVotingStates(final AbstractDataStore datastore, final String shardName,
final SimpleEntry<String, Boolean>... expStates) throws Exception {
@SafeVarargs
private static void verifyVotingStates(final AbstractDataStore datastore, final String shardName,
final SimpleEntry<String, Boolean>... expStates) throws Exception {
- String localMemberName = datastore.getActorContext().getCurrentMemberName().getName();
+ String localMemberName = datastore.getActorUtils().getCurrentMemberName().getName();
Map<String, Boolean> expStateMap = new HashMap<>();
for (Entry<String, Boolean> e: expStates) {
expStateMap.put(ShardIdentifier.create(shardName, MemberName.forName(e.getKey()),
Map<String, Boolean> expStateMap = new HashMap<>();
for (Entry<String, Boolean> e: expStates) {
expStateMap.put(ShardIdentifier.create(shardName, MemberName.forName(e.getKey()),
- datastore.getActorContext().getDataStoreName()).toString(), e.getValue());
+ datastore.getActorUtils().getDataStoreName()).toString(), e.getValue());
}
verifyRaftState(datastore, shardName, raftState -> {
String localPeerId = ShardIdentifier.create(shardName, MemberName.forName(localMemberName),
}
verifyRaftState(datastore, shardName, raftState -> {
String localPeerId = ShardIdentifier.create(shardName, MemberName.forName(localMemberName),
- datastore.getActorContext().getDataStoreName()).toString();
+ datastore.getActorUtils().getDataStoreName()).toString();
assertEquals("Voting state for " + localPeerId, expStateMap.get(localPeerId), raftState.isVoting());
for (Entry<String, Boolean> e: raftState.getPeerVotingStates().entrySet()) {
assertEquals("Voting state for " + e.getKey(), expStateMap.get(e.getKey()), e.getValue());
assertEquals("Voting state for " + localPeerId, expStateMap.get(localPeerId), raftState.isVoting());
for (Entry<String, Boolean> e: raftState.getPeerVotingStates().entrySet()) {
assertEquals("Voting state for " + e.getKey(), expStateMap.get(e.getKey()), e.getValue());
import org.opendaylight.controller.cluster.datastore.DatastoreContextFactory;
import org.opendaylight.controller.cluster.datastore.config.Configuration;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.DatastoreContextFactory;
import org.opendaylight.controller.cluster.datastore.config.Configuration;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
- ClientBackedDataStore(final ActorContext actorContext, final ClientIdentifier identifier,
+ ClientBackedDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier,
final DataStoreClient clientActor) {
final DataStoreClient clientActor) {
- super(actorContext, identifier, clientActor);
+ super(actorUtils, identifier, clientActor);
}
private boolean debugAllocation() {
}
private boolean debugAllocation() {
- return getActorContext().getDatastoreContext().isTransactionDebugContextEnabled();
+ return getActorUtils().getDatastoreContext().isTransactionDebugContextEnabled();
}
private Throwable allocationContext() {
}
private Throwable allocationContext() {
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
+import static com.google.common.base.Verify.verifyNotNull;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorRef;
import akka.util.Timeout;
import akka.actor.ActorRef;
import akka.util.Timeout;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Verify;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
import org.opendaylight.controller.cluster.access.client.AbstractClientActor;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
import org.opendaylight.controller.cluster.access.client.AbstractClientActor;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.common.actor.ExplicitAsk;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.common.actor.ExplicitAsk;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import scala.Function1;
import scala.concurrent.Await;
import scala.concurrent.duration.Duration;
import scala.Function1;
import scala.concurrent.Await;
import scala.concurrent.duration.Duration;
public abstract class AbstractDataStoreClientActor extends AbstractClientActor {
private static final Function1<ActorRef, ?> GET_CLIENT_FACTORY = ExplicitAsk.toScala(GetClientRequest::new);
public abstract class AbstractDataStoreClientActor extends AbstractClientActor {
private static final Function1<ActorRef, ?> GET_CLIENT_FACTORY = ExplicitAsk.toScala(GetClientRequest::new);
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
- AbstractDataStoreClientActor(final FrontendIdentifier frontendId, final ActorContext actorContext) {
+ AbstractDataStoreClientActor(final FrontendIdentifier frontendId, final ActorUtils actorUtils) {
- this.actorContext = Preconditions.checkNotNull(actorContext);
+ this.actorUtils = requireNonNull(actorUtils);
}
@Override
protected ClientActorConfig getClientActorConfig() {
}
@Override
protected ClientActorConfig getClientActorConfig() {
- return actorContext.getDatastoreContext();
+ return actorUtils.getDatastoreContext();
}
@Override
protected final AbstractDataStoreClientBehavior initialBehavior(final ClientActorContext context) {
}
@Override
protected final AbstractDataStoreClientBehavior initialBehavior(final ClientActorContext context) {
- return Verify.verifyNotNull(initialBehavior(context, actorContext));
+ return verifyNotNull(initialBehavior(context, actorUtils));
}
@SuppressWarnings("checkstyle:hiddenField")
}
@SuppressWarnings("checkstyle:hiddenField")
- abstract AbstractDataStoreClientBehavior initialBehavior(ClientActorContext context, ActorContext actorContext);
+ abstract AbstractDataStoreClientBehavior initialBehavior(ClientActorContext context, ActorUtils actorUtils);
@SuppressWarnings("checkstyle:IllegalCatch")
public static DataStoreClient getDistributedDataStoreClient(@Nonnull final ActorRef actor,
@SuppressWarnings("checkstyle:IllegalCatch")
public static DataStoreClient getDistributedDataStoreClient(@Nonnull final ActorRef actor,
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorRef;
import akka.util.Timeout;
import akka.actor.ActorRef;
import akka.util.Timeout;
-import com.google.common.base.Preconditions;
import com.google.common.primitives.UnsignedLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import com.google.common.primitives.UnsignedLong;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.yangtools.concepts.Registration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.opendaylight.yangtools.concepts.Registration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
* shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
/**
* {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
* shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
- * shard leader is resolved via {@link ActorContext}. The product of resolution is {@link ShardBackendInfo}.
+ * shard leader is resolved via {@link ActorUtils}. The product of resolution is {@link ShardBackendInfo}.
*
* @author Robert Varga
*/
*
* @author Robert Varga
*/
private ShardBackendInfo result;
ShardState(final CompletionStage<ShardBackendInfo> stage) {
private ShardBackendInfo result;
ShardState(final CompletionStage<ShardBackendInfo> stage) {
- this.stage = Preconditions.checkNotNull(stage);
+ this.stage = requireNonNull(stage);
stage.whenComplete(this::onStageResolved);
}
stage.whenComplete(this::onStageResolved);
}
private synchronized void onStageResolved(final ShardBackendInfo info, final Throwable failure) {
if (failure == null) {
private synchronized void onStageResolved(final ShardBackendInfo info, final Throwable failure) {
if (failure == null) {
- this.result = Preconditions.checkNotNull(info);
+ this.result = requireNonNull(info);
} else {
LOG.warn("Failed to resolve shard", failure);
}
} else {
LOG.warn("Failed to resolve shard", failure);
}
private final AtomicLong nextSessionId = new AtomicLong();
private final Function1<ActorRef, ?> connectFunction;
private final AtomicLong nextSessionId = new AtomicLong();
private final Function1<ActorRef, ?> connectFunction;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final Set<Consumer<Long>> staleBackendInfoCallbacks = ConcurrentHashMap.newKeySet();
// FIXME: we really need just ActorContext.findPrimaryShardAsync()
private final Set<Consumer<Long>> staleBackendInfoCallbacks = ConcurrentHashMap.newKeySet();
// FIXME: we really need just ActorContext.findPrimaryShardAsync()
- AbstractShardBackendResolver(final ClientIdentifier clientId, final ActorContext actorContext) {
- this.actorContext = Preconditions.checkNotNull(actorContext);
+ AbstractShardBackendResolver(final ClientIdentifier clientId, final ActorUtils actorUtils) {
+ this.actorUtils = requireNonNull(actorUtils);
this.connectFunction = ExplicitAsk.toScala(t -> new ConnectClientRequest(clientId, t, ABIVersion.BORON,
ABIVersion.current()));
}
this.connectFunction = ExplicitAsk.toScala(t -> new ConnectClientRequest(clientId, t, ABIVersion.BORON,
ABIVersion.current()));
}
staleBackendInfoCallbacks.forEach(callback -> callback.accept(cookie));
}
staleBackendInfoCallbacks.forEach(callback -> callback.accept(cookie));
}
- protected ActorContext actorContext() {
- return actorContext;
+ protected ActorUtils actorUtils() {
+ return actorUtils;
}
protected final void flushCache(final String shardName) {
}
protected final void flushCache(final String shardName) {
- actorContext.getPrimaryShardInfoCache().remove(shardName);
+ actorUtils.getPrimaryShardInfoCache().remove(shardName);
}
protected final ShardState resolveBackendInfo(final String shardName, final long cookie) {
LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);
final CompletableFuture<ShardBackendInfo> future = new CompletableFuture<>();
}
protected final ShardState resolveBackendInfo(final String shardName, final long cookie) {
LOG.debug("Resolving cookie {} to shard {}", cookie, shardName);
final CompletableFuture<ShardBackendInfo> future = new CompletableFuture<>();
- FutureConverters.toJava(actorContext.findPrimaryShardAsync(shardName)).whenComplete((info, failure) -> {
+ FutureConverters.toJava(actorUtils.findPrimaryShardAsync(shardName)).whenComplete((info, failure) -> {
if (failure == null) {
connectShard(shardName, cookie, info, future);
return;
if (failure == null) {
connectShard(shardName, cookie, info, future);
return;
private static TimeoutException wrap(final String message, final Throwable cause) {
final TimeoutException ret = new TimeoutException(message);
private static TimeoutException wrap(final String message, final Throwable cause) {
final TimeoutException ret = new TimeoutException(message);
- ret.initCause(Preconditions.checkNotNull(cause));
+ ret.initCause(requireNonNull(cause));
}
LOG.debug("Resolved backend information to {}", response);
}
LOG.debug("Resolved backend information to {}", response);
- Preconditions.checkArgument(response instanceof ConnectClientSuccess, "Unhandled response %s",
- response);
+ checkArgument(response instanceof ConnectClientSuccess, "Unhandled response %s", response);
final ConnectClientSuccess success = (ConnectClientSuccess) response;
future.complete(new ShardBackendInfo(success.getBackend(), nextSessionId.getAndIncrement(),
success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie), success.getDataTree(),
final ConnectClientSuccess success = (ConnectClientSuccess) response;
future.complete(new ShardBackendInfo(success.getBackend(), nextSessionId.getAndIncrement(),
success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie), success.getDataTree(),
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.access.concepts.FrontendType;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.access.concepts.FrontendType;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
/**
* A {@link AbstractClientActor} which acts as the point of contact for DistributedDataStore.
/**
* A {@link AbstractClientActor} which acts as the point of contact for DistributedDataStore.
* @author Robert Varga
*/
public final class DistributedDataStoreClientActor extends AbstractDataStoreClientActor {
* @author Robert Varga
*/
public final class DistributedDataStoreClientActor extends AbstractDataStoreClientActor {
- private DistributedDataStoreClientActor(final FrontendIdentifier frontendId, final ActorContext actorContext) {
- super(frontendId, actorContext);
+ private DistributedDataStoreClientActor(final FrontendIdentifier frontendId, final ActorUtils actorUtils) {
+ super(frontendId, actorUtils);
- AbstractDataStoreClientBehavior initialBehavior(final ClientActorContext context, final ActorContext actorContext) {
- return new DistributedDataStoreClientBehavior(context, actorContext);
+ AbstractDataStoreClientBehavior initialBehavior(final ClientActorContext context, final ActorUtils actorUtils) {
+ return new DistributedDataStoreClientBehavior(context, actorUtils);
}
public static Props props(@Nonnull final MemberName memberName, @Nonnull final String storeName,
}
public static Props props(@Nonnull final MemberName memberName, @Nonnull final String storeName,
- final ActorContext ctx) {
+ final ActorUtils ctx) {
final String name = "datastore-" + storeName;
final FrontendIdentifier frontendId = FrontendIdentifier.create(memberName, FrontendType.forName(name));
return Props.create(DistributedDataStoreClientActor.class,
final String name = "datastore-" + storeName;
final FrontendIdentifier frontendId = FrontendIdentifier.create(memberName, FrontendType.forName(name));
return Props.create(DistributedDataStoreClientActor.class,
import java.util.function.Function;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import java.util.function.Function;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
/**
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
/**
pathToShard = resolver::resolveShardForPath;
}
pathToShard = resolver::resolveShardForPath;
}
- DistributedDataStoreClientBehavior(final ClientActorContext context, final ActorContext actorContext) {
- this(context, new ModuleShardBackendResolver(context.getIdentifier(), actorContext));
+ DistributedDataStoreClientBehavior(final ClientActorContext context, final ActorUtils actorUtils) {
+ this(context, new ModuleShardBackendResolver(context.getIdentifier(), actorUtils));
import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
import org.opendaylight.controller.cluster.datastore.shardmanager.RegisterForShardAvailabilityChanges;
import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
import org.opendaylight.controller.cluster.datastore.shardmanager.RegisterForShardAvailabilityChanges;
import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.yangtools.concepts.Registration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.slf4j.Logger;
import org.opendaylight.yangtools.concepts.Registration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.slf4j.Logger;
/**
* {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
* shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
/**
* {@link BackendInfoResolver} implementation for static shard configuration based on ShardManager. Each string-named
* shard is assigned a single cookie and this mapping is stored in a bidirectional map. Information about corresponding
- * shard leader is resolved via {@link ActorContext}. The product of resolution is {@link ShardBackendInfo}.
+ * shard leader is resolved via {@link ActorUtils}. The product of resolution is {@link ShardBackendInfo}.
*
* @author Robert Varga
*/
*
* @author Robert Varga
*/
private volatile BiMap<String, Long> shards = ImmutableBiMap.of(DefaultShardStrategy.DEFAULT_SHARD, 0L);
// FIXME: we really need just ActorContext.findPrimaryShardAsync()
private volatile BiMap<String, Long> shards = ImmutableBiMap.of(DefaultShardStrategy.DEFAULT_SHARD, 0L);
// FIXME: we really need just ActorContext.findPrimaryShardAsync()
- ModuleShardBackendResolver(final ClientIdentifier clientId, final ActorContext actorContext) {
- super(clientId, actorContext);
+ ModuleShardBackendResolver(final ClientIdentifier clientId, final ActorUtils actorUtils) {
+ super(clientId, actorUtils);
- shardAvailabilityChangesRegFuture = ask(actorContext.getShardManager(), new RegisterForShardAvailabilityChanges(
+ shardAvailabilityChangesRegFuture = ask(actorUtils.getShardManager(), new RegisterForShardAvailabilityChanges(
this::onShardAvailabilityChange), Timeout.apply(60, TimeUnit.MINUTES))
.map(reply -> (Registration)reply, ExecutionContexts.global());
this::onShardAvailabilityChange), Timeout.apply(60, TimeUnit.MINUTES))
.map(reply -> (Registration)reply, ExecutionContexts.global());
}
Long resolveShardForPath(final YangInstanceIdentifier path) {
}
Long resolveShardForPath(final YangInstanceIdentifier path) {
- final String shardName = actorContext().getShardStrategyFactory().getStrategy(path).findShard(path);
+ final String shardName = actorUtils().getShardStrategyFactory().getStrategy(path).findShard(path);
Long cookie = shards.get(shardName);
if (cookie == null) {
synchronized (this) {
Long cookie = shards.get(shardName);
if (cookie == null) {
synchronized (this) {
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
+import static java.util.Objects.requireNonNull;
+
-import com.google.common.base.Preconditions;
import javax.annotation.Nonnull;
import org.opendaylight.controller.cluster.access.client.AbstractClientActor;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.access.concepts.FrontendType;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import javax.annotation.Nonnull;
import org.opendaylight.controller.cluster.access.client.AbstractClientActor;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.access.concepts.FrontendType;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
/**
* A {@link AbstractClientActor} which acts as the point of contact for DistributedDataStore.
/**
* A {@link AbstractClientActor} which acts as the point of contact for DistributedDataStore.
public final class SimpleDataStoreClientActor extends AbstractDataStoreClientActor {
private final String shardName;
public final class SimpleDataStoreClientActor extends AbstractDataStoreClientActor {
private final String shardName;
- private SimpleDataStoreClientActor(final FrontendIdentifier frontendId, final ActorContext actorContext,
+ private SimpleDataStoreClientActor(final FrontendIdentifier frontendId, final ActorUtils actorUtils,
final String shardName) {
final String shardName) {
- super(frontendId, actorContext);
- this.shardName = Preconditions.checkNotNull(shardName);
+ super(frontendId, actorUtils);
+ this.shardName = requireNonNull(shardName);
- AbstractDataStoreClientBehavior initialBehavior(final ClientActorContext context, final ActorContext actorContext) {
- return new SimpleDataStoreClientBehavior(context, actorContext, shardName);
+ AbstractDataStoreClientBehavior initialBehavior(final ClientActorContext context, final ActorUtils actorUtils) {
+ return new SimpleDataStoreClientBehavior(context, actorUtils, shardName);
}
public static Props props(@Nonnull final MemberName memberName, @Nonnull final String storeName,
}
public static Props props(@Nonnull final MemberName memberName, @Nonnull final String storeName,
- final ActorContext ctx, final String shardName) {
+ final ActorUtils actorUtils, final String shardName) {
final String name = "datastore-" + storeName;
final FrontendIdentifier frontendId = FrontendIdentifier.create(memberName, FrontendType.forName(name));
return Props.create(SimpleDataStoreClientActor.class,
final String name = "datastore-" + storeName;
final FrontendIdentifier frontendId = FrontendIdentifier.create(memberName, FrontendType.forName(name));
return Props.create(SimpleDataStoreClientActor.class,
- () -> new SimpleDataStoreClientActor(frontendId, ctx, shardName));
+ () -> new SimpleDataStoreClientActor(frontendId, actorUtils, shardName));
package org.opendaylight.controller.cluster.databroker.actors.dds;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
package org.opendaylight.controller.cluster.databroker.actors.dds;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
/**
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
/**
super(context, resolver);
}
super(context, resolver);
}
- SimpleDataStoreClientBehavior(final ClientActorContext context, final ActorContext actorContext,
+ SimpleDataStoreClientBehavior(final ClientActorContext context, final ActorUtils actorUtils,
final String shardName) {
final String shardName) {
- this(context, new SimpleShardBackendResolver(context.getIdentifier(), actorContext, shardName));
+ this(context, new SimpleShardBackendResolver(context.getIdentifier(), actorUtils, shardName));
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
*/
package org.opendaylight.controller.cluster.databroker.actors.dds;
-import com.google.common.base.Preconditions;
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
import java.util.concurrent.CompletionStage;
import javax.annotation.concurrent.ThreadSafe;
import org.opendaylight.controller.cluster.access.client.BackendInfoResolver;
import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
import java.util.concurrent.CompletionStage;
import javax.annotation.concurrent.ThreadSafe;
import org.opendaylight.controller.cluster.access.client.BackendInfoResolver;
import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
private volatile ShardState state;
// FIXME: we really need just ActorContext.findPrimaryShardAsync()
private volatile ShardState state;
// FIXME: we really need just ActorContext.findPrimaryShardAsync()
- SimpleShardBackendResolver(final ClientIdentifier clientId, final ActorContext actorContext,
+ SimpleShardBackendResolver(final ClientIdentifier clientId, final ActorUtils actorUtils,
final String shardName) {
final String shardName) {
- super(clientId, actorContext);
- this.shardName = Preconditions.checkNotNull(shardName);
+ super(clientId, actorUtils);
+ this.shardName = requireNonNull(shardName);
}
private CompletionStage<ShardBackendInfo> getBackendInfo(final long cookie) {
}
private CompletionStage<ShardBackendInfo> getBackendInfo(final long cookie) {
- Preconditions.checkArgument(cookie == 0);
+ checkArgument(cookie == 0);
final ShardState existing = state;
if (existing != null) {
final ShardState existing = state;
if (existing != null) {
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
package org.opendaylight.controller.cluster.datastore;
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.PoisonPill;
import akka.actor.Props;
import com.google.common.annotations.VisibleForTesting;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.PoisonPill;
import akka.actor.Props;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.Uninterruptibles;
import java.util.concurrent.CountDownLatch;
import com.google.common.base.Throwables;
import com.google.common.util.concurrent.Uninterruptibles;
import java.util.concurrent.CountDownLatch;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.DatastoreInfoMXBeanImpl;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManagerCreator;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.DatastoreInfoMXBeanImpl;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.shardmanager.ShardManagerCreator;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
private static final long READY_WAIT_FACTOR = 3;
private static final long READY_WAIT_FACTOR = 3;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final long waitTillReadyTimeInMillis;
private AutoCloseable closeable;
private final long waitTillReadyTimeInMillis;
private AutoCloseable closeable;
protected AbstractDataStore(final ActorSystem actorSystem, final ClusterWrapper cluster,
final Configuration configuration, final DatastoreContextFactory datastoreContextFactory,
final DatastoreSnapshot restoreFromSnapshot) {
protected AbstractDataStore(final ActorSystem actorSystem, final ClusterWrapper cluster,
final Configuration configuration, final DatastoreContextFactory datastoreContextFactory,
final DatastoreSnapshot restoreFromSnapshot) {
- Preconditions.checkNotNull(actorSystem, "actorSystem should not be null");
- Preconditions.checkNotNull(cluster, "cluster should not be null");
- Preconditions.checkNotNull(configuration, "configuration should not be null");
- Preconditions.checkNotNull(datastoreContextFactory, "datastoreContextFactory should not be null");
+ requireNonNull(actorSystem, "actorSystem should not be null");
+ requireNonNull(cluster, "cluster should not be null");
+ requireNonNull(configuration, "configuration should not be null");
+ requireNonNull(datastoreContextFactory, "datastoreContextFactory should not be null");
String shardManagerId = ShardManagerIdentifier.builder()
.type(datastoreContextFactory.getBaseDatastoreContext().getDataStoreName()).build().toString();
String shardManagerId = ShardManagerIdentifier.builder()
.type(datastoreContextFactory.getBaseDatastoreContext().getDataStoreName()).build().toString();
.restoreFromSnapshot(restoreFromSnapshot)
.distributedDataStore(this);
.restoreFromSnapshot(restoreFromSnapshot)
.distributedDataStore(this);
- actorContext = new ActorContext(actorSystem, createShardManager(actorSystem, creator, shardDispatcher,
+ actorUtils = new ActorUtils(actorSystem, createShardManager(actorSystem, creator, shardDispatcher,
shardManagerId), cluster, configuration, datastoreContextFactory.getBaseDatastoreContext(),
primaryShardInfoCache);
final Props clientProps = DistributedDataStoreClientActor.props(cluster.getCurrentMemberName(),
shardManagerId), cluster, configuration, datastoreContextFactory.getBaseDatastoreContext(),
primaryShardInfoCache);
final Props clientProps = DistributedDataStoreClientActor.props(cluster.getCurrentMemberName(),
- datastoreContextFactory.getBaseDatastoreContext().getDataStoreName(), actorContext);
+ datastoreContextFactory.getBaseDatastoreContext().getDataStoreName(), actorUtils);
final ActorRef clientActor = actorSystem.actorOf(clientProps);
try {
client = DistributedDataStoreClientActor.getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
final ActorRef clientActor = actorSystem.actorOf(clientProps);
try {
client = DistributedDataStoreClientActor.getDistributedDataStoreClient(clientActor, 30, TimeUnit.SECONDS);
identifier = client.getIdentifier();
LOG.debug("Distributed data store client {} started", identifier);
identifier = client.getIdentifier();
LOG.debug("Distributed data store client {} started", identifier);
- this.waitTillReadyTimeInMillis = actorContext.getDatastoreContext().getShardLeaderElectionTimeout()
+ this.waitTillReadyTimeInMillis = actorUtils.getDatastoreContext().getShardLeaderElectionTimeout()
.duration().toMillis() * READY_WAIT_FACTOR;
datastoreConfigMXBean = new DatastoreConfigurationMXBeanImpl(
.duration().toMillis() * READY_WAIT_FACTOR;
datastoreConfigMXBean = new DatastoreConfigurationMXBeanImpl(
datastoreConfigMXBean.registerMBean();
datastoreInfoMXBean = new DatastoreInfoMXBeanImpl(datastoreContextFactory.getBaseDatastoreContext()
datastoreConfigMXBean.registerMBean();
datastoreInfoMXBean = new DatastoreInfoMXBeanImpl(datastoreContextFactory.getBaseDatastoreContext()
- .getDataStoreMXBeanType(), actorContext);
+ .getDataStoreMXBeanType(), actorUtils);
datastoreInfoMXBean.registerMBean();
}
@VisibleForTesting
datastoreInfoMXBean.registerMBean();
}
@VisibleForTesting
- protected AbstractDataStore(final ActorContext actorContext, final ClientIdentifier identifier) {
- this.actorContext = Preconditions.checkNotNull(actorContext, "actorContext should not be null");
+ protected AbstractDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier) {
+ this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null");
- this.identifier = Preconditions.checkNotNull(identifier);
- this.waitTillReadyTimeInMillis = actorContext.getDatastoreContext().getShardLeaderElectionTimeout()
+ this.identifier = requireNonNull(identifier);
+ this.waitTillReadyTimeInMillis = actorUtils.getDatastoreContext().getShardLeaderElectionTimeout()
.duration().toMillis() * READY_WAIT_FACTOR;
}
@VisibleForTesting
.duration().toMillis() * READY_WAIT_FACTOR;
}
@VisibleForTesting
- protected AbstractDataStore(final ActorContext actorContext, final ClientIdentifier identifier,
+ protected AbstractDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier,
final DataStoreClient clientActor) {
final DataStoreClient clientActor) {
- this.actorContext = Preconditions.checkNotNull(actorContext, "actorContext should not be null");
+ this.actorUtils = requireNonNull(actorUtils, "actorContext should not be null");
this.client = clientActor;
this.client = clientActor;
- this.identifier = Preconditions.checkNotNull(identifier);
- this.waitTillReadyTimeInMillis = actorContext.getDatastoreContext().getShardLeaderElectionTimeout()
+ this.identifier = requireNonNull(identifier);
+ this.waitTillReadyTimeInMillis = actorUtils.getDatastoreContext().getShardLeaderElectionTimeout()
.duration().toMillis() * READY_WAIT_FACTOR;
}
.duration().toMillis() * READY_WAIT_FACTOR;
}
@Override
public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerTreeChangeListener(
final YangInstanceIdentifier treeId, final L listener) {
@Override
public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerTreeChangeListener(
final YangInstanceIdentifier treeId, final L listener) {
- Preconditions.checkNotNull(treeId, "treeId should not be null");
- Preconditions.checkNotNull(listener, "listener should not be null");
+ requireNonNull(treeId, "treeId should not be null");
+ requireNonNull(listener, "listener should not be null");
- final String shardName = actorContext.getShardStrategyFactory().getStrategy(treeId).findShard(treeId);
+ final String shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId);
LOG.debug("Registering tree listener: {} for tree: {} shard: {}", listener, treeId, shardName);
final DataTreeChangeListenerProxy<L> listenerRegistrationProxy =
LOG.debug("Registering tree listener: {} for tree: {} shard: {}", listener, treeId, shardName);
final DataTreeChangeListenerProxy<L> listenerRegistrationProxy =
- new DataTreeChangeListenerProxy<>(actorContext, listener, treeId);
+ new DataTreeChangeListenerProxy<>(actorUtils, listener, treeId);
listenerRegistrationProxy.init(shardName);
return listenerRegistrationProxy;
listenerRegistrationProxy.init(shardName);
return listenerRegistrationProxy;
@Override
public <C extends DOMDataTreeCommitCohort> DOMDataTreeCommitCohortRegistration<C> registerCommitCohort(
final DOMDataTreeIdentifier subtree, final C cohort) {
@Override
public <C extends DOMDataTreeCommitCohort> DOMDataTreeCommitCohortRegistration<C> registerCommitCohort(
final DOMDataTreeIdentifier subtree, final C cohort) {
- YangInstanceIdentifier treeId =
- Preconditions.checkNotNull(subtree, "subtree should not be null").getRootIdentifier();
- Preconditions.checkNotNull(cohort, "listener should not be null");
+ YangInstanceIdentifier treeId = requireNonNull(subtree, "subtree should not be null").getRootIdentifier();
+ requireNonNull(cohort, "listener should not be null");
- final String shardName = actorContext.getShardStrategyFactory().getStrategy(treeId).findShard(treeId);
+ final String shardName = actorUtils.getShardStrategyFactory().getStrategy(treeId).findShard(treeId);
LOG.debug("Registering cohort: {} for tree: {} shard: {}", cohort, treeId, shardName);
DataTreeCohortRegistrationProxy<C> cohortProxy =
LOG.debug("Registering cohort: {} for tree: {} shard: {}", cohort, treeId, shardName);
DataTreeCohortRegistrationProxy<C> cohortProxy =
- new DataTreeCohortRegistrationProxy<>(actorContext, subtree, cohort);
+ new DataTreeCohortRegistrationProxy<>(actorUtils, subtree, cohort);
cohortProxy.init(shardName);
return cohortProxy;
}
@Override
public void onGlobalContextUpdated(final SchemaContext schemaContext) {
cohortProxy.init(shardName);
return cohortProxy;
}
@Override
public void onGlobalContextUpdated(final SchemaContext schemaContext) {
- actorContext.setSchemaContext(schemaContext);
+ actorUtils.setSchemaContext(schemaContext);
}
@Override
public void onDatastoreContextUpdated(final DatastoreContextFactory contextFactory) {
}
@Override
public void onDatastoreContextUpdated(final DatastoreContextFactory contextFactory) {
- LOG.info("DatastoreContext updated for data store {}", actorContext.getDataStoreName());
+ LOG.info("DatastoreContext updated for data store {}", actorUtils.getDataStoreName());
- actorContext.setDatastoreContext(contextFactory);
+ actorUtils.setDatastoreContext(contextFactory);
datastoreConfigMXBean.setContext(contextFactory.getBaseDatastoreContext());
}
datastoreConfigMXBean.setContext(contextFactory.getBaseDatastoreContext());
}
- actorContext.shutdown();
if (client != null) {
client.close();
if (client != null) {
client.close();
- public ActorContext getActorContext() {
- return actorContext;
+ public ActorUtils getActorUtils() {
+ return actorUtils;
}
public void waitTillReady() {
}
public void waitTillReady() {
final YangInstanceIdentifier insideShard,
final org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener delegate) {
final YangInstanceIdentifier insideShard,
final org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener delegate) {
- Preconditions.checkNotNull(shardLookup, "shardLookup should not be null");
- Preconditions.checkNotNull(insideShard, "insideShard should not be null");
- Preconditions.checkNotNull(delegate, "delegate should not be null");
+ requireNonNull(shardLookup, "shardLookup should not be null");
+ requireNonNull(insideShard, "insideShard should not be null");
+ requireNonNull(delegate, "delegate should not be null");
- final String shardName = actorContext.getShardStrategyFactory().getStrategy(shardLookup).findShard(shardLookup);
+ final String shardName = actorUtils.getShardStrategyFactory().getStrategy(shardLookup).findShard(shardLookup);
LOG.debug("Registering tree listener: {} for tree: {} shard: {}, path inside shard: {}",
delegate,shardLookup, shardName, insideShard);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> listenerRegistrationProxy =
LOG.debug("Registering tree listener: {} for tree: {} shard: {}, path inside shard: {}",
delegate,shardLookup, shardName, insideShard);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> listenerRegistrationProxy =
- new DataTreeChangeListenerProxy<>(actorContext,
+ new DataTreeChangeListenerProxy<>(actorUtils,
// wrap this in the ClusteredDOMDataTreeChangeLister interface
// since we always want clustered registration
(ClusteredDOMDataTreeChangeListener) delegate::onDataTreeChanged, insideShard);
// wrap this in the ClusteredDOMDataTreeChangeLister interface
// since we always want clustered registration
(ClusteredDOMDataTreeChangeListener) delegate::onDataTreeChanged, insideShard);
public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerShardConfigListener(
final YangInstanceIdentifier internalPath,
final DOMDataTreeChangeListener delegate) {
public <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerShardConfigListener(
final YangInstanceIdentifier internalPath,
final DOMDataTreeChangeListener delegate) {
- Preconditions.checkNotNull(delegate, "delegate should not be null");
+ requireNonNull(delegate, "delegate should not be null");
LOG.debug("Registering a listener for the configuration shard: {}", internalPath);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
LOG.debug("Registering a listener for the configuration shard: {}", internalPath);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy =
- new DataTreeChangeListenerProxy<>(actorContext, delegate, internalPath);
+ new DataTreeChangeListenerProxy<>(actorUtils, delegate, internalPath);
proxy.init(ClusterUtils.PREFIX_CONFIG_SHARD_ID);
return (ListenerRegistration<L>) proxy;
proxy.init(ClusterUtils.PREFIX_CONFIG_SHARD_ID);
return (ListenerRegistration<L>) proxy;
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.dispatch.OnComplete;
import akka.actor.ActorSelection;
import akka.dispatch.OnComplete;
-import com.google.common.base.Preconditions;
import java.util.Collection;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.Collection;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
private final ConcurrentMap<String, F> knownLocal = new ConcurrentHashMap<>();
private final LocalHistoryIdentifier historyId;
private final ConcurrentMap<String, F> knownLocal = new ConcurrentHashMap<>();
private final LocalHistoryIdentifier historyId;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
// Used via TX_COUNTER_UPDATER
@SuppressWarnings("unused")
private volatile long nextTx;
// Used via TX_COUNTER_UPDATER
@SuppressWarnings("unused")
private volatile long nextTx;
- protected AbstractTransactionContextFactory(final ActorContext actorContext,
- final LocalHistoryIdentifier historyId) {
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.historyId = Preconditions.checkNotNull(historyId);
+ protected AbstractTransactionContextFactory(final ActorUtils actorUtils, final LocalHistoryIdentifier historyId) {
+ this.actorUtils = requireNonNull(actorUtils);
+ this.historyId = requireNonNull(historyId);
- final ActorContext getActorContext() {
- return actorContext;
+ final ActorUtils getActorUtils() {
+ return actorUtils;
}
final LocalHistoryIdentifier getHistoryId() {
}
final LocalHistoryIdentifier getHistoryId() {
final TransactionContextWrapper newTransactionContextWrapper(final TransactionProxy parent,
final String shardName) {
final TransactionContextWrapper transactionContextWrapper =
final TransactionContextWrapper newTransactionContextWrapper(final TransactionProxy parent,
final String shardName) {
final TransactionContextWrapper transactionContextWrapper =
- new TransactionContextWrapper(parent.getIdentifier(), actorContext, shardName);
+ new TransactionContextWrapper(parent.getIdentifier(), actorUtils, shardName);
Future<PrimaryShardInfo> findPrimaryFuture = findPrimaryShard(shardName, parent.getIdentifier());
if (findPrimaryFuture.isCompleted()) {
Future<PrimaryShardInfo> findPrimaryFuture = findPrimaryShard(shardName, parent.getIdentifier());
if (findPrimaryFuture.isCompleted()) {
onFindPrimaryShardFailure(failure, parent, shardName, transactionContextWrapper);
}
}
onFindPrimaryShardFailure(failure, parent, shardName, transactionContextWrapper);
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
return transactionContextWrapper;
}
return transactionContextWrapper;
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.PoisonPill;
import akka.dispatch.OnComplete;
import com.google.common.annotations.VisibleForTesting;
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.PoisonPill;
import akka.dispatch.OnComplete;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistration;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
import org.opendaylight.controller.cluster.datastore.messages.CloseDataTreeNotificationListenerRegistration;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
final class DataTreeChangeListenerProxy<T extends DOMDataTreeChangeListener> extends AbstractListenerRegistration<T> {
private static final Logger LOG = LoggerFactory.getLogger(DataTreeChangeListenerProxy.class);
private final ActorRef dataChangeListenerActor;
final class DataTreeChangeListenerProxy<T extends DOMDataTreeChangeListener> extends AbstractListenerRegistration<T> {
private static final Logger LOG = LoggerFactory.getLogger(DataTreeChangeListenerProxy.class);
private final ActorRef dataChangeListenerActor;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final YangInstanceIdentifier registeredPath;
@GuardedBy("this")
private ActorSelection listenerRegistrationActor;
private final YangInstanceIdentifier registeredPath;
@GuardedBy("this")
private ActorSelection listenerRegistrationActor;
- DataTreeChangeListenerProxy(final ActorContext actorContext, final T listener,
+ DataTreeChangeListenerProxy(final ActorUtils actorUtils, final T listener,
final YangInstanceIdentifier registeredPath) {
super(listener);
final YangInstanceIdentifier registeredPath) {
super(listener);
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.registeredPath = Preconditions.checkNotNull(registeredPath);
- this.dataChangeListenerActor = actorContext.getActorSystem().actorOf(
+ this.actorUtils = requireNonNull(actorUtils);
+ this.registeredPath = requireNonNull(registeredPath);
+ this.dataChangeListenerActor = actorUtils.getActorSystem().actorOf(
DataTreeChangeListenerActor.props(getInstance(), registeredPath)
DataTreeChangeListenerActor.props(getInstance(), registeredPath)
- .withDispatcher(actorContext.getNotificationDispatcherPath()));
+ .withDispatcher(actorUtils.getNotificationDispatcherPath()));
- LOG.debug("{}: Created actor {} for DTCL {}", actorContext.getDatastoreContext().getLogicalStoreType(),
+ LOG.debug("{}: Created actor {} for DTCL {}", actorUtils.getDatastoreContext().getLogicalStoreType(),
dataChangeListenerActor, listener);
}
dataChangeListenerActor, listener);
}
}
void init(final String shardName) {
}
void init(final String shardName) {
- Future<ActorRef> findFuture = actorContext.findLocalShardAsync(shardName);
+ Future<ActorRef> findFuture = actorUtils.findLocalShardAsync(shardName);
findFuture.onComplete(new OnComplete<ActorRef>() {
@Override
public void onComplete(final Throwable failure, final ActorRef shard) {
findFuture.onComplete(new OnComplete<ActorRef>() {
@Override
public void onComplete(final Throwable failure, final ActorRef shard) {
doRegistration(shard);
}
}
doRegistration(shard);
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
private void setListenerRegistrationActor(final ActorSelection actor) {
}
private void setListenerRegistrationActor(final ActorSelection actor) {
private void doRegistration(final ActorRef shard) {
private void doRegistration(final ActorRef shard) {
- Future<Object> future = actorContext.executeOperationAsync(shard,
+ Future<Object> future = actorUtils.executeOperationAsync(shard,
new RegisterDataTreeChangeListener(registeredPath, dataChangeListenerActor,
getInstance() instanceof ClusteredDOMDataTreeChangeListener),
new RegisterDataTreeChangeListener(registeredPath, dataChangeListenerActor,
getInstance() instanceof ClusteredDOMDataTreeChangeListener),
- actorContext.getDatastoreContext().getShardInitializationTimeout());
+ actorUtils.getDatastoreContext().getShardInitializationTimeout());
future.onComplete(new OnComplete<Object>() {
@Override
future.onComplete(new OnComplete<Object>() {
@Override
getInstance(), registeredPath, failure);
} else {
RegisterDataTreeNotificationListenerReply reply = (RegisterDataTreeNotificationListenerReply)result;
getInstance(), registeredPath, failure);
} else {
RegisterDataTreeNotificationListenerReply reply = (RegisterDataTreeNotificationListenerReply)result;
- setListenerRegistrationActor(actorContext.actorSelection(
+ setListenerRegistrationActor(actorUtils.actorSelection(
reply.getListenerRegistrationPath()));
}
}
reply.getListenerRegistrationPath()));
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
private String logContext() {
}
private String logContext() {
- return actorContext.getDatastoreContext().getLogicalStoreType().toString();
+ return actorUtils.getDatastoreContext().getLogicalStoreType().toString();
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorRef;
import akka.dispatch.OnComplete;
import akka.pattern.Patterns;
import akka.util.Timeout;
import akka.actor.ActorRef;
import akka.dispatch.OnComplete;
import akka.pattern.Patterns;
import akka.util.Timeout;
-import com.google.common.base.Preconditions;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistration;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistration;
import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
private static final Timeout TIMEOUT = new Timeout(new FiniteDuration(5, TimeUnit.SECONDS));
private final DOMDataTreeIdentifier subtree;
private final ActorRef actor;
private static final Timeout TIMEOUT = new Timeout(new FiniteDuration(5, TimeUnit.SECONDS));
private final DOMDataTreeIdentifier subtree;
private final ActorRef actor;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
@GuardedBy("this")
private ActorRef cohortRegistry;
@GuardedBy("this")
private ActorRef cohortRegistry;
- DataTreeCohortRegistrationProxy(final ActorContext actorContext, final DOMDataTreeIdentifier subtree,
+ DataTreeCohortRegistrationProxy(final ActorUtils actorUtils, final DOMDataTreeIdentifier subtree,
final C cohort) {
super(cohort);
final C cohort) {
super(cohort);
- this.subtree = Preconditions.checkNotNull(subtree);
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.actor = actorContext.getActorSystem().actorOf(DataTreeCohortActor.props(getInstance(),
- subtree.getRootIdentifier()).withDispatcher(actorContext.getNotificationDispatcherPath()));
+ this.subtree = requireNonNull(subtree);
+ this.actorUtils = requireNonNull(actorUtils);
+ this.actor = actorUtils.getActorSystem().actorOf(DataTreeCohortActor.props(getInstance(),
+ subtree.getRootIdentifier()).withDispatcher(actorUtils.getNotificationDispatcherPath()));
}
public void init(final String shardName) {
// FIXME: Add late binding to shard.
}
public void init(final String shardName) {
// FIXME: Add late binding to shard.
- Future<ActorRef> findFuture = actorContext.findLocalShardAsync(shardName);
+ Future<ActorRef> findFuture = actorUtils.findLocalShardAsync(shardName);
findFuture.onComplete(new OnComplete<ActorRef>() {
@Override
public void onComplete(final Throwable failure, final ActorRef shard) {
findFuture.onComplete(new OnComplete<ActorRef>() {
@Override
public void onComplete(final Throwable failure, final ActorRef shard) {
performRegistration(shard);
}
}
performRegistration(shard);
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
private synchronized void performRegistration(final ActorRef shard) {
}
private synchronized void performRegistration(final ActorRef shard) {
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
import org.opendaylight.controller.cluster.datastore.config.Configuration;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
import org.opendaylight.controller.cluster.datastore.config.Configuration;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
final Configuration configuration, final DatastoreContextFactory datastoreContextFactory,
final DatastoreSnapshot restoreFromSnapshot) {
super(actorSystem, cluster, configuration, datastoreContextFactory, restoreFromSnapshot);
final Configuration configuration, final DatastoreContextFactory datastoreContextFactory,
final DatastoreSnapshot restoreFromSnapshot) {
super(actorSystem, cluster, configuration, datastoreContextFactory, restoreFromSnapshot);
- this.txContextFactory = new TransactionContextFactory(getActorContext(), getIdentifier());
+ this.txContextFactory = new TransactionContextFactory(getActorUtils(), getIdentifier());
- DistributedDataStore(final ActorContext actorContext, final ClientIdentifier identifier) {
- super(actorContext, identifier);
- this.txContextFactory = new TransactionContextFactory(getActorContext(), getIdentifier());
+ DistributedDataStore(final ActorUtils actorUtils, final ClientIdentifier identifier) {
+ super(actorUtils, identifier);
+ this.txContextFactory = new TransactionContextFactory(getActorUtils(), getIdentifier());
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
- getActorContext().acquireTxCreationPermit();
+ getActorUtils().acquireTxCreationPermit();
return new TransactionProxy(txContextFactory, TransactionType.WRITE_ONLY);
}
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
return new TransactionProxy(txContextFactory, TransactionType.WRITE_ONLY);
}
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
- getActorContext().acquireTxCreationPermit();
+ getActorUtils().acquireTxCreationPermit();
return new TransactionProxy(txContextFactory, TransactionType.READ_WRITE);
}
return new TransactionProxy(txContextFactory, TransactionType.READ_WRITE);
}
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStore;
/**
import org.opendaylight.mdsal.dom.spi.store.DOMStore;
/**
*/
public interface DistributedDataStoreInterface extends DOMStore {
*/
public interface DistributedDataStoreInterface extends DOMStore {
- ActorContext getActorContext();
+ ActorUtils getActorUtils();
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.dispatch.Futures;
import akka.dispatch.OnComplete;
import akka.actor.ActorSelection;
import akka.dispatch.Futures;
import akka.dispatch.OnComplete;
-import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.Optional;
import java.util.SortedSet;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.Optional;
import java.util.SortedSet;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.mdsal.dom.spi.store.SnapshotBackedWriteTransaction;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.mdsal.dom.spi.store.SnapshotBackedWriteTransaction;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
private final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction;
private final DataTreeModification modification;
private final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction;
private final DataTreeModification modification;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final ActorSelection leader;
private final Exception operationError;
private final ActorSelection leader;
private final Exception operationError;
- protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
+ protected LocalThreePhaseCommitCohort(final ActorUtils actorUtils, final ActorSelection leader,
final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
final DataTreeModification modification,
final Exception operationError) {
final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
final DataTreeModification modification,
final Exception operationError) {
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.leader = Preconditions.checkNotNull(leader);
- this.transaction = Preconditions.checkNotNull(transaction);
- this.modification = Preconditions.checkNotNull(modification);
+ this.actorUtils = requireNonNull(actorUtils);
+ this.leader = requireNonNull(leader);
+ this.transaction = requireNonNull(transaction);
+ this.modification = requireNonNull(modification);
this.operationError = operationError;
}
this.operationError = operationError;
}
- protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
+ protected LocalThreePhaseCommitCohort(final ActorUtils actorUtils, final ActorSelection leader,
final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction, final Exception operationError) {
final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction, final Exception operationError) {
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.leader = Preconditions.checkNotNull(leader);
- this.transaction = Preconditions.checkNotNull(transaction);
- this.operationError = Preconditions.checkNotNull(operationError);
+ this.actorUtils = requireNonNull(actorUtils);
+ this.leader = requireNonNull(leader);
+ this.transaction = requireNonNull(transaction);
+ this.operationError = requireNonNull(operationError);
this.modification = null;
}
this.modification = null;
}
final ReadyLocalTransaction message = new ReadyLocalTransaction(transaction.getIdentifier(),
modification, immediate, participatingShardNames);
final ReadyLocalTransaction message = new ReadyLocalTransaction(transaction.getIdentifier(),
modification, immediate, participatingShardNames);
- return actorContext.executeOperationAsync(leader, message, actorContext.getTransactionCommitOperationTimeout());
+ return actorUtils.executeOperationAsync(leader, message, actorUtils.getTransactionCommitOperationTimeout());
}
Future<ActorSelection> initiateCoordinatedCommit(final Optional<SortedSet<String>> participatingShardNames) {
final Future<Object> messageFuture = initiateCommit(false, participatingShardNames);
}
Future<ActorSelection> initiateCoordinatedCommit(final Optional<SortedSet<String>> participatingShardNames) {
final Future<Object> messageFuture = initiateCommit(false, participatingShardNames);
- final Future<ActorSelection> ret = TransactionReadyReplyMapper.transform(messageFuture, actorContext,
+ final Future<ActorSelection> ret = TransactionReadyReplyMapper.transform(messageFuture, actorUtils,
transaction.getIdentifier());
ret.onComplete(new OnComplete<ActorSelection>() {
@Override
transaction.getIdentifier());
ret.onComplete(new OnComplete<ActorSelection>() {
@Override
LOG.debug("Transaction {} resolved to actor {}", transaction.getIdentifier(), success);
}
LOG.debug("Transaction {} resolved to actor {}", transaction.getIdentifier(), success);
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
transactionAborted(transaction);
}
}
transactionAborted(transaction);
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
protected LocalChainThreePhaseCommitCohort(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
DataTreeModification modification, Exception operationError) {
protected LocalChainThreePhaseCommitCohort(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
DataTreeModification modification, Exception operationError) {
- super(parent.getActorContext(), leader, transaction, modification, operationError);
+ super(parent.getActorUtils(), leader, transaction, modification, operationError);
}
protected LocalChainThreePhaseCommitCohort(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
Exception operationError) {
}
protected LocalChainThreePhaseCommitCohort(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
Exception operationError) {
- super(parent.getActorContext(), leader, transaction, operationError);
+ super(parent.getActorUtils(), leader, transaction, operationError);
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.actor.ActorSelection;
-import com.google.common.base.Preconditions;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
private final ActorSelection leader;
private final DataTree dataTree;
private final ActorSelection leader;
private final DataTree dataTree;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
- LocalTransactionFactoryImpl(final ActorContext actorContext, final ActorSelection leader, final DataTree dataTree) {
- this.leader = Preconditions.checkNotNull(leader);
- this.dataTree = Preconditions.checkNotNull(dataTree);
- this.actorContext = actorContext;
+ LocalTransactionFactoryImpl(final ActorUtils actorUtils, final ActorSelection leader, final DataTree dataTree) {
+ this.leader = requireNonNull(leader);
+ this.dataTree = requireNonNull(dataTree);
+ this.actorUtils = actorUtils;
}
DataTree getDataTree() {
}
DataTree getDataTree() {
final SnapshotBackedWriteTransaction<TransactionIdentifier> tx,
final DataTreeModification tree,
final Exception readyError) {
final SnapshotBackedWriteTransaction<TransactionIdentifier> tx,
final DataTreeModification tree,
final Exception readyError) {
- return new LocalThreePhaseCommitCohort(actorContext, leader, tx, tree, readyError);
+ return new LocalThreePhaseCommitCohort(actorUtils, leader, tx, tree, readyError);
}
@SuppressWarnings({"unchecked", "checkstyle:IllegalCatch"})
@Override
public LocalThreePhaseCommitCohort onTransactionReady(@Nonnull DOMStoreWriteTransaction tx,
@Nullable Exception operationError) {
}
@SuppressWarnings({"unchecked", "checkstyle:IllegalCatch"})
@Override
public LocalThreePhaseCommitCohort onTransactionReady(@Nonnull DOMStoreWriteTransaction tx,
@Nullable Exception operationError) {
- Preconditions.checkArgument(tx instanceof SnapshotBackedWriteTransaction);
+ checkArgument(tx instanceof SnapshotBackedWriteTransaction);
if (operationError != null) {
if (operationError != null) {
- return new LocalThreePhaseCommitCohort(actorContext, leader,
+ return new LocalThreePhaseCommitCohort(actorUtils, leader,
(SnapshotBackedWriteTransaction<TransactionIdentifier>)tx, operationError);
}
(SnapshotBackedWriteTransaction<TransactionIdentifier>)tx, operationError);
}
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.dispatch.Futures;
import akka.dispatch.OnComplete;
import akka.actor.ActorSelection;
import akka.dispatch.Futures;
import akka.dispatch.OnComplete;
-import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.SettableFuture;
import java.util.Optional;
import java.util.SortedSet;
import com.google.common.util.concurrent.SettableFuture;
import java.util.Optional;
import java.util.SortedSet;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
import org.opendaylight.controller.cluster.datastore.modification.AbstractModification;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
import org.opendaylight.controller.cluster.datastore.modification.AbstractModification;
import org.opendaylight.controller.cluster.datastore.modification.Modification;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.common.api.ReadFailedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.opendaylight.mdsal.common.api.ReadFailedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class RemoteTransactionContext extends AbstractTransactionContext {
private static final Logger LOG = LoggerFactory.getLogger(RemoteTransactionContext.class);
public class RemoteTransactionContext extends AbstractTransactionContext {
private static final Logger LOG = LoggerFactory.getLogger(RemoteTransactionContext.class);
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final ActorSelection actor;
private final OperationLimiter limiter;
private final ActorSelection actor;
private final OperationLimiter limiter;
private volatile Throwable failedModification;
protected RemoteTransactionContext(final TransactionIdentifier identifier, final ActorSelection actor,
private volatile Throwable failedModification;
protected RemoteTransactionContext(final TransactionIdentifier identifier, final ActorSelection actor,
- final ActorContext actorContext, final short remoteTransactionVersion, final OperationLimiter limiter) {
+ final ActorUtils actorUtils, final short remoteTransactionVersion, final OperationLimiter limiter) {
super(identifier, remoteTransactionVersion);
super(identifier, remoteTransactionVersion);
- this.limiter = Preconditions.checkNotNull(limiter);
+ this.limiter = requireNonNull(limiter);
- this.actorContext = actorContext;
+ this.actorUtils = actorUtils;
}
private ActorSelection getActor() {
return actor;
}
}
private ActorSelection getActor() {
return actor;
}
- protected ActorContext getActorContext() {
- return actorContext;
+ protected ActorUtils getActorUtils() {
+ return actorUtils;
LOG.debug("Tx {} closeTransaction called", getIdentifier());
TransactionContextCleanup.untrack(this);
LOG.debug("Tx {} closeTransaction called", getIdentifier());
TransactionContextCleanup.untrack(this);
- actorContext.sendOperationAsync(getActor(), new CloseTransaction(getTransactionVersion()).toSerializable());
+ actorUtils.sendOperationAsync(getActor(), new CloseTransaction(getTransactionVersion()).toSerializable());
// Transform the last reply Future into a Future that returns the cohort actor path from
// the last reply message. That's the end result of the ready operation.
// Transform the last reply Future into a Future that returns the cohort actor path from
// the last reply message. That's the end result of the ready operation.
- return TransactionReadyReplyMapper.transform(readyReplyFuture, actorContext, getIdentifier());
+ return TransactionReadyReplyMapper.transform(readyReplyFuture, actorUtils, getIdentifier());
}
private BatchedModifications newBatchedModifications() {
}
private BatchedModifications newBatchedModifications() {
batchedModifications.addModification(modification);
if (batchedModifications.getModifications().size()
batchedModifications.addModification(modification);
if (batchedModifications.getModifications().size()
- >= actorContext.getDatastoreContext().getShardBatchedModificationCount()) {
+ >= actorUtils.getDatastoreContext().getShardBatchedModificationCount()) {
sendBatchedModifications();
}
}
sendBatchedModifications();
}
}
- sent = actorContext.executeOperationAsync(getActor(), toSend.toSerializable(),
- actorContext.getTransactionCommitOperationTimeout());
+ sent = actorUtils.executeOperationAsync(getActor(), toSend.toSerializable(),
+ actorUtils.getTransactionCommitOperationTimeout());
sent.onComplete(new OnComplete<Object>() {
@Override
public void onComplete(final Throwable failure, final Object success) {
sent.onComplete(new OnComplete<Object>() {
@Override
public void onComplete(final Throwable failure, final Object success) {
}
limiter.release(permitsToRelease);
}
}
limiter.release(permitsToRelease);
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
- final Future<Object> future = actorContext.executeOperationAsync(getActor(),
- readCmd.asVersion(getTransactionVersion()).toSerializable(), actorContext.getOperationTimeout());
- future.onComplete(onComplete, actorContext.getClientDispatcher());
+ final Future<Object> future = actorUtils.executeOperationAsync(getActor(),
+ readCmd.asVersion(getTransactionVersion()).toSerializable(), actorUtils.getOperationTimeout());
+ future.onComplete(onComplete, actorUtils.getClientDispatcher());
* @return True if a permit was successfully acquired, false otherwise
*/
private boolean acquireOperation() {
* @return True if a permit was successfully acquired, false otherwise
*/
private boolean acquireOperation() {
- Preconditions.checkState(isOperationHandOffComplete(),
+ checkState(isOperationHandOffComplete(),
"Attempted to acquire execute operation permit for transaction %s on actor %s during handoff",
getIdentifier(), actor);
"Attempted to acquire execute operation permit for transaction %s on actor %s during handoff",
getIdentifier(), actor);
import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
// For the total create tx timeout, use 2 times the election timeout. This should be enough time for
// a leader re-election to occur if we happen to hit it in transition.
// For the total create tx timeout, use 2 times the election timeout. This should be enough time for
// a leader re-election to occur if we happen to hit it in transition.
- totalCreateTxTimeout = parent.getActorContext().getDatastoreContext().getShardRaftConfig()
+ totalCreateTxTimeout = parent.getActorUtils().getDatastoreContext().getShardRaftConfig()
.getElectionTimeOutInterval().toMillis() * 2;
// We'll use the operationTimeout for the the create Tx message timeout so it can be set appropriately
// for unit tests but cap it at MAX_CREATE_TX_MSG_TIMEOUT_IN_MS. The operationTimeout could be set
// larger than the totalCreateTxTimeout in production which we don't want.
.getElectionTimeOutInterval().toMillis() * 2;
// We'll use the operationTimeout for the the create Tx message timeout so it can be set appropriately
// for unit tests but cap it at MAX_CREATE_TX_MSG_TIMEOUT_IN_MS. The operationTimeout could be set
// larger than the totalCreateTxTimeout in production which we don't want.
- long operationTimeout = parent.getActorContext().getOperationTimeout().duration().toMillis();
+ long operationTimeout = parent.getActorUtils().getOperationTimeout().duration().toMillis();
createTxMessageTimeout = new Timeout(Math.min(operationTimeout, MAX_CREATE_TX_MSG_TIMEOUT_IN_MS),
TimeUnit.MILLISECONDS);
}
createTxMessageTimeout = new Timeout(Math.min(operationTimeout, MAX_CREATE_TX_MSG_TIMEOUT_IN_MS),
TimeUnit.MILLISECONDS);
}
return parent.getType();
}
return parent.getType();
}
- private ActorContext getActorContext() {
- return parent.getActorContext();
+ private ActorUtils getActorUtils() {
+ return parent.getActorUtils();
}
private TransactionIdentifier getIdentifier() {
}
private TransactionIdentifier getIdentifier() {
this.primaryShardInfo = newPrimaryShardInfo;
if (getTransactionType() == TransactionType.WRITE_ONLY
this.primaryShardInfo = newPrimaryShardInfo;
if (getTransactionType() == TransactionType.WRITE_ONLY
- && getActorContext().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
+ && getActorUtils().getDatastoreContext().isWriteOnlyTransactionOptimizationsEnabled()) {
ActorSelection primaryShard = newPrimaryShardInfo.getPrimaryShardActor();
LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
ActorSelection primaryShard = newPrimaryShardInfo.getPrimaryShardActor();
LOG.debug("Tx {} Primary shard {} found - creating WRITE_ONLY transaction context",
Object serializedCreateMessage = new CreateTransaction(getIdentifier(), getTransactionType().ordinal(),
primaryShardInfo.getPrimaryShardVersion()).toSerializable();
Object serializedCreateMessage = new CreateTransaction(getIdentifier(), getTransactionType().ordinal(),
primaryShardInfo.getPrimaryShardVersion()).toSerializable();
- Future<Object> createTxFuture = getActorContext().executeOperationAsync(
+ Future<Object> createTxFuture = getActorUtils().executeOperationAsync(
primaryShardInfo.getPrimaryShardActor(), serializedCreateMessage, createTxMessageTimeout);
createTxFuture.onComplete(new OnComplete<Object>() {
primaryShardInfo.getPrimaryShardActor(), serializedCreateMessage, createTxMessageTimeout);
createTxFuture.onComplete(new OnComplete<Object>() {
public void onComplete(final Throwable failure, final Object response) {
onCreateTransactionComplete(failure, response);
}
public void onComplete(final Throwable failure, final Object response) {
onCreateTransactionComplete(failure, response);
}
- }, getActorContext().getClientDispatcher());
+ }, getActorUtils().getClientDispatcher());
}
private void tryFindPrimaryShard() {
LOG.debug("Tx {} Retrying findPrimaryShardAsync for shard {}", getIdentifier(), shardName);
this.primaryShardInfo = null;
}
private void tryFindPrimaryShard() {
LOG.debug("Tx {} Retrying findPrimaryShardAsync for shard {}", getIdentifier(), shardName);
this.primaryShardInfo = null;
- Future<PrimaryShardInfo> findPrimaryFuture = getActorContext().findPrimaryShardAsync(shardName);
+ Future<PrimaryShardInfo> findPrimaryFuture = getActorUtils().findPrimaryShardAsync(shardName);
findPrimaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
@Override
public void onComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) {
onFindPrimaryShardComplete(failure, newPrimaryShardInfo);
}
findPrimaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
@Override
public void onComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) {
onFindPrimaryShardComplete(failure, newPrimaryShardInfo);
}
- }, getActorContext().getClientDispatcher());
+ }, getActorUtils().getClientDispatcher());
}
private void onFindPrimaryShardComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) {
}
private void onFindPrimaryShardComplete(final Throwable failure, final PrimaryShardInfo newPrimaryShardInfo) {
LOG.debug("Tx {}: create tx on shard {} failed with exception \"{}\" - scheduling retry in {} ms",
getIdentifier(), shardName, failure, scheduleInterval);
LOG.debug("Tx {}: create tx on shard {} failed with exception \"{}\" - scheduling retry in {} ms",
getIdentifier(), shardName, failure, scheduleInterval);
- getActorContext().getActorSystem().scheduler().scheduleOnce(
+ getActorUtils().getActorSystem().scheduler().scheduleOnce(
FiniteDuration.create(scheduleInterval, TimeUnit.MILLISECONDS),
FiniteDuration.create(scheduleInterval, TimeUnit.MILLISECONDS),
- this::tryFindPrimaryShard, getActorContext().getClientDispatcher());
+ this::tryFindPrimaryShard, getActorUtils().getClientDispatcher());
private TransactionContext createValidTransactionContext(final CreateTransactionReply reply) {
LOG.debug("Tx {} Received {}", getIdentifier(), reply);
private TransactionContext createValidTransactionContext(final CreateTransactionReply reply) {
LOG.debug("Tx {} Received {}", getIdentifier(), reply);
- return createValidTransactionContext(getActorContext().actorSelection(reply.getTransactionPath()),
+ return createValidTransactionContext(getActorUtils().actorSelection(reply.getTransactionPath()),
reply.getTransactionPath(), primaryShardInfo.getPrimaryShardVersion());
}
private TransactionContext createValidTransactionContext(final ActorSelection transactionActor,
final String transactionPath, final short remoteTransactionVersion) {
final TransactionContext ret = new RemoteTransactionContext(transactionContextWrapper.getIdentifier(),
reply.getTransactionPath(), primaryShardInfo.getPrimaryShardVersion());
}
private TransactionContext createValidTransactionContext(final ActorSelection transactionActor,
final String transactionPath, final short remoteTransactionVersion) {
final TransactionContext ret = new RemoteTransactionContext(transactionContextWrapper.getIdentifier(),
- transactionActor, getActorContext(), remoteTransactionVersion, transactionContextWrapper.getLimiter());
+ transactionActor, getActorUtils(), remoteTransactionVersion, transactionContextWrapper.getLimiter());
if (parent.getType() == TransactionType.READ_ONLY) {
TransactionContextCleanup.track(parent, ret);
if (parent.getType() == TransactionType.READ_ONLY) {
TransactionContextCleanup.track(parent, ret);
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.dispatch.OnComplete;
import akka.dispatch.OnComplete;
-import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.Arrays;
import java.util.List;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import java.util.Arrays;
import java.util.List;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
class SingleCommitCohortProxy extends AbstractThreePhaseCommitCohort<Object> {
private static final Logger LOG = LoggerFactory.getLogger(SingleCommitCohortProxy.class);
class SingleCommitCohortProxy extends AbstractThreePhaseCommitCohort<Object> {
private static final Logger LOG = LoggerFactory.getLogger(SingleCommitCohortProxy.class);
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final Future<Object> cohortFuture;
private final TransactionIdentifier transactionId;
private volatile DOMStoreThreePhaseCommitCohort delegateCohort = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
private final OperationCallback.Reference operationCallbackRef;
private final Future<Object> cohortFuture;
private final TransactionIdentifier transactionId;
private volatile DOMStoreThreePhaseCommitCohort delegateCohort = NoOpDOMStoreThreePhaseCommitCohort.INSTANCE;
private final OperationCallback.Reference operationCallbackRef;
- SingleCommitCohortProxy(ActorContext actorContext, Future<Object> cohortFuture, TransactionIdentifier transactionId,
+ SingleCommitCohortProxy(ActorUtils actorUtils, Future<Object> cohortFuture, TransactionIdentifier transactionId,
OperationCallback.Reference operationCallbackRef) {
OperationCallback.Reference operationCallbackRef) {
- this.actorContext = actorContext;
+ this.actorUtils = actorUtils;
this.cohortFuture = cohortFuture;
this.cohortFuture = cohortFuture;
- this.transactionId = Preconditions.checkNotNull(transactionId);
+ this.transactionId = requireNonNull(transactionId);
this.operationCallbackRef = operationCallbackRef;
}
this.operationCallbackRef = operationCallbackRef;
}
// immediate success, to complete the 3PC for the front-end.
returnFuture.set(Boolean.TRUE);
}
// immediate success, to complete the 3PC for the front-end.
returnFuture.set(Boolean.TRUE);
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
* terms of the Eclipse Public License v1.0 which accompanies this distribution,
* and is available at http://www.eclipse.org/legal/epl-v10.html
*/
package org.opendaylight.controller.cluster.datastore;
package org.opendaylight.controller.cluster.datastore;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.dispatch.OnComplete;
import akka.actor.ActorSelection;
import akka.dispatch.OnComplete;
-import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.base.Supplier;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.FutureCallback;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final List<CohortInfo> cohorts;
private final SettableFuture<Void> cohortsResolvedFuture = SettableFuture.create();
private final TransactionIdentifier transactionId;
private volatile OperationCallback commitOperationCallback;
private final List<CohortInfo> cohorts;
private final SettableFuture<Void> cohortsResolvedFuture = SettableFuture.create();
private final TransactionIdentifier transactionId;
private volatile OperationCallback commitOperationCallback;
- public ThreePhaseCommitCohortProxy(final ActorContext actorContext, final List<CohortInfo> cohorts,
+ public ThreePhaseCommitCohortProxy(final ActorUtils actorUtils, final List<CohortInfo> cohorts,
final TransactionIdentifier transactionId) {
final TransactionIdentifier transactionId) {
- this.actorContext = actorContext;
+ this.actorUtils = actorUtils;
- this.transactionId = Preconditions.checkNotNull(transactionId);
+ this.transactionId = requireNonNull(transactionId);
if (cohorts.isEmpty()) {
cohortsResolvedFuture.set(null);
if (cohorts.isEmpty()) {
cohortsResolvedFuture.set(null);
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
return cohortsResolvedFuture;
}
return cohortsResolvedFuture;
- commitOperationCallback = new TransactionRateLimitingCallback(actorContext);
+ commitOperationCallback = new TransactionRateLimitingCallback(actorUtils);
commitOperationCallback.run();
final Iterator<CohortInfo> iterator = cohorts.iterator();
commitOperationCallback.run();
final Iterator<CohortInfo> iterator = cohorts.iterator();
LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor());
LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor());
- Future<Object> future = actorContext.executeOperationAsync(toCohortInfo.getResolvedActor(),
- message.toSerializable(), actorContext.getTransactionCommitOperationTimeout());
- future.onComplete(onComplete, actorContext.getClientDispatcher());
+ Future<Object> future = actorUtils.executeOperationAsync(toCohortInfo.getResolvedActor(),
+ message.toSerializable(), actorUtils.getTransactionCommitOperationTimeout());
+ future.onComplete(onComplete, actorUtils.getClientDispatcher());
}
private Future<Iterable<Object>> invokeCohorts(final MessageSupplier messageSupplier) {
}
private Future<Iterable<Object>> invokeCohorts(final MessageSupplier messageSupplier) {
LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort.getResolvedActor());
LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort.getResolvedActor());
- futureList.add(actorContext.executeOperationAsync(cohort.getResolvedActor(), message,
- actorContext.getTransactionCommitOperationTimeout()));
+ futureList.add(actorUtils.executeOperationAsync(cohort.getResolvedActor(), message,
+ actorUtils.getTransactionCommitOperationTimeout()));
- return akka.dispatch.Futures.sequence(futureList, actorContext.getClientDispatcher());
+ return akka.dispatch.Futures.sequence(futureList, actorUtils.getClientDispatcher());
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
short getActorVersion() {
}
short getActorVersion() {
- Preconditions.checkState(resolvedActor != null,
- "getActorVersion cannot be called until the actor is resolved");
+ checkState(resolvedActor != null, "getActorVersion cannot be called until the actor is resolved");
return actorVersionSupplier.get();
}
}
return actorVersionSupplier.get();
}
}
new ConcurrentHashMap<>();
TransactionChainProxy(final TransactionContextFactory parent, final LocalHistoryIdentifier historyId) {
new ConcurrentHashMap<>();
TransactionChainProxy(final TransactionContextFactory parent, final LocalHistoryIdentifier historyId) {
- super(parent.getActorContext(), historyId);
+ super(parent.getActorUtils(), historyId);
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
@Override
public DOMStoreReadWriteTransaction newReadWriteTransaction() {
- getActorContext().acquireTxCreationPermit();
+ getActorUtils().acquireTxCreationPermit();
return allocateWriteTransaction(TransactionType.READ_WRITE);
}
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
return allocateWriteTransaction(TransactionType.READ_WRITE);
}
@Override
public DOMStoreWriteTransaction newWriteOnlyTransaction() {
- getActorContext().acquireTxCreationPermit();
+ getActorUtils().acquireTxCreationPermit();
return allocateWriteTransaction(TransactionType.WRITE_ONLY);
}
return allocateWriteTransaction(TransactionType.WRITE_ONLY);
}
// Send a close transaction chain request to each and every shard
// Send a close transaction chain request to each and every shard
- getActorContext().broadcast(version -> new CloseTransactionChain(getHistoryId(), version).toSerializable(),
+ getActorUtils().broadcast(version -> new CloseTransactionChain(getHistoryId(), version).toSerializable(),
CloseTransactionChain.class);
}
CloseTransactionChain.class);
}
- previous.onComplete(onComplete, getActorContext().getClientDispatcher());
+ previous.onComplete(onComplete, getActorUtils().getClientDispatcher());
return returnPromise.future();
}
return returnPromise.future();
}
}
Future<Iterable<Object>> combinedFutures = Futures.sequence(priorReadOnlyTxFutures,
}
Future<Iterable<Object>> combinedFutures = Futures.sequence(priorReadOnlyTxFutures,
- getActorContext().getClientDispatcher());
+ getActorUtils().getClientDispatcher());
final Promise<T> returnPromise = Futures.promise();
final OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
final Promise<T> returnPromise = Futures.promise();
final OnComplete<Iterable<Object>> onComplete = new OnComplete<Iterable<Object>>() {
- combinedFutures.onComplete(onComplete, getActorContext().getClientDispatcher());
+ combinedFutures.onComplete(onComplete, getActorUtils().getClientDispatcher());
return returnPromise.future();
} else {
return future;
return returnPromise.future();
} else {
return future;
}
// Combine the ready Futures into 1
}
// Combine the ready Futures into 1
- final Future<Iterable<T>> combined = Futures.sequence(cohortFutures, getActorContext().getClientDispatcher());
+ final Future<Iterable<T>> combined = Futures.sequence(cohortFutures, getActorUtils().getClientDispatcher());
// Record the we have outstanding futures
final State newState = new Submitted(transaction, combined);
// Record the we have outstanding futures
final State newState = new Submitted(transaction, combined);
public void onComplete(final Throwable arg0, final Iterable<T> arg1) {
STATE_UPDATER.compareAndSet(TransactionChainProxy.this, newState, IDLE_STATE);
}
public void onComplete(final Throwable arg0, final Iterable<T> arg1) {
STATE_UPDATER.compareAndSet(TransactionChainProxy.this, newState, IDLE_STATE);
}
- }, getActorContext().getClientDispatcher());
+ }, getActorUtils().getClientDispatcher());
import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import scala.concurrent.Future;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import scala.concurrent.Future;
final class TransactionContextFactory extends AbstractTransactionContextFactory<LocalTransactionFactoryImpl> {
private final AtomicLong nextHistory = new AtomicLong(1);
final class TransactionContextFactory extends AbstractTransactionContextFactory<LocalTransactionFactoryImpl> {
private final AtomicLong nextHistory = new AtomicLong(1);
- TransactionContextFactory(final ActorContext actorContext, final ClientIdentifier clientId) {
- super(actorContext, new LocalHistoryIdentifier(clientId, 0));
+ TransactionContextFactory(final ActorUtils actorUtils, final ClientIdentifier clientId) {
+ super(actorUtils, new LocalHistoryIdentifier(clientId, 0));
@Override
protected LocalTransactionFactoryImpl factoryForShard(final String shardName, final ActorSelection shardLeader,
final DataTree dataTree) {
@Override
protected LocalTransactionFactoryImpl factoryForShard(final String shardName, final ActorSelection shardLeader,
final DataTree dataTree) {
- return new LocalTransactionFactoryImpl(getActorContext(), shardLeader, dataTree);
+ return new LocalTransactionFactoryImpl(getActorUtils(), shardLeader, dataTree);
}
@Override
protected Future<PrimaryShardInfo> findPrimaryShard(final String shardName, TransactionIdentifier txId) {
}
@Override
protected Future<PrimaryShardInfo> findPrimaryShard(final String shardName, TransactionIdentifier txId) {
- return getActorContext().findPrimaryShardAsync(shardName);
+ return getActorUtils().findPrimaryShardAsync(shardName);
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.dispatch.Futures;
import akka.actor.ActorSelection;
import akka.dispatch.Futures;
-import com.google.common.base.Preconditions;
import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.ArrayList;
import java.util.Collection;
import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
@GuardedBy("queuedTxOperations")
private boolean pendingEnqueue;
@GuardedBy("queuedTxOperations")
private boolean pendingEnqueue;
- TransactionContextWrapper(final TransactionIdentifier identifier, final ActorContext actorContext,
+ TransactionContextWrapper(final TransactionIdentifier identifier, final ActorUtils actorUtils,
final String shardName) {
final String shardName) {
- this.identifier = Preconditions.checkNotNull(identifier);
+ this.identifier = requireNonNull(identifier);
this.limiter = new OperationLimiter(identifier,
// 1 extra permit for the ready operation
this.limiter = new OperationLimiter(identifier,
// 1 extra permit for the ready operation
- actorContext.getDatastoreContext().getShardBatchedModificationCount() + 1,
- TimeUnit.MILLISECONDS.toSeconds(actorContext.getDatastoreContext().getOperationTimeoutInMillis()));
- this.shardName = Preconditions.checkNotNull(shardName);
+ actorUtils.getDatastoreContext().getShardBatchedModificationCount() + 1,
+ TimeUnit.MILLISECONDS.toSeconds(actorUtils.getDatastoreContext().getOperationTimeoutInMillis()));
+ this.shardName = requireNonNull(shardName);
}
TransactionContext getTransactionContext() {
}
TransactionContext getTransactionContext() {
synchronized (queuedTxOperations) {
contextOnEntry = transactionContext;
if (contextOnEntry == null) {
synchronized (queuedTxOperations) {
contextOnEntry = transactionContext;
if (contextOnEntry == null) {
- Preconditions.checkState(pendingEnqueue == false, "Concurrent access to transaction %s detected",
- identifier);
+ checkState(pendingEnqueue == false, "Concurrent access to transaction %s detected", identifier);
pendingEnqueue = true;
}
}
pendingEnqueue = true;
}
}
import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator;
import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator;
import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
@VisibleForTesting
public TransactionProxy(final AbstractTransactionContextFactory<?> txContextFactory, final TransactionType type) {
@VisibleForTesting
public TransactionProxy(final AbstractTransactionContextFactory<?> txContextFactory, final TransactionType type) {
- super(txContextFactory.nextIdentifier(), txContextFactory.getActorContext().getDatastoreContext()
+ super(txContextFactory.nextIdentifier(), txContextFactory.getActorUtils().getDatastoreContext()
.isTransactionDebugContextEnabled());
this.txContextFactory = txContextFactory;
this.type = Preconditions.checkNotNull(type);
.isTransactionDebugContextEnabled());
this.txContextFactory = txContextFactory;
this.type = Preconditions.checkNotNull(type);
}
private FluentFuture<Optional<NormalizedNode<?, ?>>> readAllData() {
}
private FluentFuture<Optional<NormalizedNode<?, ?>>> readAllData() {
- final Set<String> allShardNames = txContextFactory.getActorContext().getConfiguration().getAllShardNames();
+ final Set<String> allShardNames = txContextFactory.getActorUtils().getConfiguration().getAllShardNames();
final Collection<FluentFuture<Optional<NormalizedNode<?, ?>>>> futures = new ArrayList<>(allShardNames.size());
for (String shardName : allShardNames) {
final Collection<FluentFuture<Optional<NormalizedNode<?, ?>>>> futures = new ArrayList<>(allShardNames.size());
for (String shardName : allShardNames) {
(Function<List<Optional<NormalizedNode<?, ?>>>, Optional<NormalizedNode<?, ?>>>) input -> {
try {
return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input,
(Function<List<Optional<NormalizedNode<?, ?>>>, Optional<NormalizedNode<?, ?>>>) input -> {
try {
return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input,
- txContextFactory.getActorContext().getSchemaContext(),
- txContextFactory.getActorContext().getDatastoreContext().getLogicalStoreType());
+ txContextFactory.getActorUtils().getSchemaContext(),
+ txContextFactory.getActorUtils().getDatastoreContext().getLogicalStoreType());
} catch (DataValidationFailedException e) {
throw new IllegalArgumentException("Failed to aggregate", e);
}
} catch (DataValidationFailedException e) {
throw new IllegalArgumentException("Failed to aggregate", e);
}
future = getDirectCommitFuture(transactionContext, operationCallbackRef, null);
}
future = getDirectCommitFuture(transactionContext, operationCallbackRef, null);
}
- return new SingleCommitCohortProxy(txContextFactory.getActorContext(), future, getIdentifier(),
+ return new SingleCommitCohortProxy(txContextFactory.getActorUtils(), future, getIdentifier(),
operationCallbackRef);
}
private Future<?> getDirectCommitFuture(final TransactionContext transactionContext,
final OperationCallback.Reference operationCallbackRef, final Boolean havePermit) {
TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback(
operationCallbackRef);
}
private Future<?> getDirectCommitFuture(final TransactionContext transactionContext,
final OperationCallback.Reference operationCallbackRef, final Boolean havePermit) {
TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback(
- txContextFactory.getActorContext());
+ txContextFactory.getActorUtils());
operationCallbackRef.set(rateLimitingCallback);
rateLimitingCallback.run();
return transactionContext.directCommit(havePermit);
operationCallbackRef.set(rateLimitingCallback);
rateLimitingCallback.run();
return transactionContext.directCommit(havePermit);
- return new ThreePhaseCommitCohortProxy(txContextFactory.getActorContext(), cohorts, getIdentifier());
+ return new ThreePhaseCommitCohortProxy(txContextFactory.getActorUtils(), cohorts, getIdentifier());
}
private String shardNameFromIdentifier(final YangInstanceIdentifier path) {
}
private String shardNameFromIdentifier(final YangInstanceIdentifier path) {
- return txContextFactory.getActorContext().getShardStrategyFactory().getStrategy(path).findShard(path);
+ return txContextFactory.getActorUtils().getShardStrategyFactory().getStrategy(path).findShard(path);
}
private TransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) {
}
private TransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) {
return state != TransactionState.OPEN;
}
return state != TransactionState.OPEN;
}
- ActorContext getActorContext() {
- return txContextFactory.getActorContext();
+ ActorUtils getActorUtils() {
+ return txContextFactory.getActorUtils();
import com.google.common.base.Preconditions;
import com.google.common.base.Ticker;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Preconditions;
import com.google.common.base.Ticker;
import java.util.concurrent.TimeUnit;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
/**
* TransactionRateLimitingCallback computes the new transaction rate limit on the successful completion of a
/**
* TransactionRateLimitingCallback computes the new transaction rate limit on the successful completion of a
private long elapsedTime;
private volatile State state = State.STOPPED;
private long elapsedTime;
private volatile State state = State.STOPPED;
- TransactionRateLimitingCallback(ActorContext actorContext) {
- commitTimer = actorContext.getOperationTimer(ActorContext.COMMIT);
+ TransactionRateLimitingCallback(ActorUtils actorUtils) {
+ commitTimer = actorUtils.getOperationTimer(ActorUtils.COMMIT);
*/
package org.opendaylight.controller.cluster.datastore;
*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.dispatch.Mapper;
import akka.actor.ActorSelection;
import akka.dispatch.Mapper;
-import com.google.common.base.Preconditions;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
};
private static final Logger LOG = LoggerFactory.getLogger(TransactionReadyReplyMapper.class);
private final TransactionIdentifier identifier;
};
private static final Logger LOG = LoggerFactory.getLogger(TransactionReadyReplyMapper.class);
private final TransactionIdentifier identifier;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
- protected TransactionReadyReplyMapper(final ActorContext actorContext, final TransactionIdentifier identifier) {
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.identifier = Preconditions.checkNotNull(identifier);
+ protected TransactionReadyReplyMapper(final ActorUtils actorUtils, final TransactionIdentifier identifier) {
+ this.actorUtils = requireNonNull(actorUtils);
+ this.identifier = requireNonNull(identifier);
- protected final ActorContext getActorContext() {
- return actorContext;
+ protected final ActorUtils getActorUtils() {
+ return actorUtils;
}
protected String extractCohortPathFrom(final ReadyTransactionReply readyTxReply) {
}
protected String extractCohortPathFrom(final ReadyTransactionReply readyTxReply) {
// actor path from the reply.
if (ReadyTransactionReply.isSerializedType(serializedReadyReply)) {
ReadyTransactionReply readyTxReply = ReadyTransactionReply.fromSerializable(serializedReadyReply);
// actor path from the reply.
if (ReadyTransactionReply.isSerializedType(serializedReadyReply)) {
ReadyTransactionReply readyTxReply = ReadyTransactionReply.fromSerializable(serializedReadyReply);
- return actorContext.actorSelection(extractCohortPathFrom(readyTxReply));
+ return actorUtils.actorSelection(extractCohortPathFrom(readyTxReply));
}
// Throwing an exception here will fail the Future.
}
// Throwing an exception here will fail the Future.
identifier, serializedReadyReply.getClass()));
}
identifier, serializedReadyReply.getClass()));
}
- static Future<ActorSelection> transform(final Future<Object> readyReplyFuture, final ActorContext actorContext,
+ static Future<ActorSelection> transform(final Future<Object> readyReplyFuture, final ActorUtils actorUtils,
final TransactionIdentifier identifier) {
final TransactionIdentifier identifier) {
- return readyReplyFuture.transform(new TransactionReadyReplyMapper(actorContext, identifier),
- SAME_FAILURE_TRANSFORMER, actorContext.getClientDispatcher());
+ return readyReplyFuture.transform(new TransactionReadyReplyMapper(actorUtils, identifier),
+ SAME_FAILURE_TRANSFORMER, actorUtils.getClientDispatcher());
import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ModuleShardStrategy;
import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
import org.opendaylight.controller.cluster.datastore.messages.GetShardDataTree;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ModuleShardStrategy;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.mdsal.eos.common.api.CandidateAlreadyRegisteredException;
import org.opendaylight.mdsal.eos.common.api.EntityOwnershipState;
import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
import org.opendaylight.mdsal.eos.common.api.CandidateAlreadyRegisteredException;
import org.opendaylight.mdsal.eos.common.api.EntityOwnershipState;
import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
private static final Timeout MESSAGE_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
private final ConcurrentMap<DOMEntity, DOMEntity> registeredEntities = new ConcurrentHashMap<>();
private static final Timeout MESSAGE_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
private final ConcurrentMap<DOMEntity, DOMEntity> registeredEntities = new ConcurrentHashMap<>();
- private final ActorContext context;
+ private final ActorUtils context;
private volatile ActorRef localEntityOwnershipShard;
private volatile DataTree localEntityOwnershipShardDataTree;
private volatile ActorRef localEntityOwnershipShard;
private volatile DataTree localEntityOwnershipShardDataTree;
- DistributedEntityOwnershipService(final ActorContext context) {
+ DistributedEntityOwnershipService(final ActorUtils context) {
this.context = Preconditions.checkNotNull(context);
}
this.context = Preconditions.checkNotNull(context);
}
- public static DistributedEntityOwnershipService start(final ActorContext context,
+ public static DistributedEntityOwnershipService start(final ActorUtils context,
final EntityOwnerSelectionStrategyConfig strategyConfig) {
ActorRef shardManagerActor = context.getShardManager();
final EntityOwnerSelectionStrategyConfig strategyConfig) {
ActorRef shardManagerActor = context.getShardManager();
- private static EntityOwnershipShard.Builder newShardBuilder(final ActorContext context,
+ private static EntityOwnershipShard.Builder newShardBuilder(final ActorUtils context,
final EntityOwnerSelectionStrategyConfig strategyConfig) {
return EntityOwnershipShard.newBuilder().localMemberName(context.getCurrentMemberName())
.ownerSelectionStrategyConfig(strategyConfig);
final EntityOwnerSelectionStrategyConfig strategyConfig) {
return EntityOwnershipShard.newBuilder().localMemberName(context.getCurrentMemberName())
.ownerSelectionStrategyConfig(strategyConfig);
*/
package org.opendaylight.controller.cluster.datastore.jmx.mbeans;
*/
package org.opendaylight.controller.cluster.datastore.jmx.mbeans;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.md.sal.common.util.jmx.AbstractMXBean;
/**
import org.opendaylight.controller.md.sal.common.util.jmx.AbstractMXBean;
/**
*/
public class DatastoreInfoMXBeanImpl extends AbstractMXBean implements DatastoreInfoMXBean {
*/
public class DatastoreInfoMXBeanImpl extends AbstractMXBean implements DatastoreInfoMXBean {
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
- public DatastoreInfoMXBeanImpl(String mxBeanType, ActorContext actorContext) {
+ public DatastoreInfoMXBeanImpl(String mxBeanType, ActorUtils actorUtils) {
super("GeneralRuntimeInfo", mxBeanType, null);
super("GeneralRuntimeInfo", mxBeanType, null);
- this.actorContext = actorContext;
+ this.actorUtils = actorUtils;
}
@Override
public double getTransactionCreationRateLimit() {
}
@Override
public double getTransactionCreationRateLimit() {
- return actorContext.getTxCreationLimit();
+ return actorUtils.getTxCreationLimit();
import scala.concurrent.duration.FiniteDuration;
/**
import scala.concurrent.duration.FiniteDuration;
/**
- * The ActorContext class contains utility methods which could be used by
- * non-actors (like DistributedDataStore) to work with actors a little more
- * easily. An ActorContext can be freely passed around to local object instances
- * but should not be passed to actors especially remote actors
+ * The ActorUtils class contains utility methods which could be used by non-actors (like DistributedDataStore) to work
+ * with actors a little more easily. An ActorContext can be freely passed around to local object instances but should
+ * not be passed to actors especially remote actors.
-public class ActorContext {
- private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class);
+public class ActorUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(ActorUtils.class);
private static final String DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store";
private static final String METRIC_RATE = "rate";
private static final Mapper<Throwable, Throwable> FIND_PRIMARY_FAILURE_TRANSFORMER =
private static final String DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store";
private static final String METRIC_RATE = "rate";
private static final Mapper<Throwable, Throwable> FIND_PRIMARY_FAILURE_TRANSFORMER =
private final PrimaryShardInfoFutureCache primaryShardInfoCache;
private final ShardStrategyFactory shardStrategyFactory;
private final PrimaryShardInfoFutureCache primaryShardInfoCache;
private final ShardStrategyFactory shardStrategyFactory;
- public ActorContext(final ActorSystem actorSystem, final ActorRef shardManager,
+ public ActorUtils(final ActorSystem actorSystem, final ActorRef shardManager,
final ClusterWrapper clusterWrapper, final Configuration configuration) {
this(actorSystem, shardManager, clusterWrapper, configuration,
DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
}
final ClusterWrapper clusterWrapper, final Configuration configuration) {
this(actorSystem, shardManager, clusterWrapper, configuration,
DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
}
- public ActorContext(final ActorSystem actorSystem, final ActorRef shardManager,
+ public ActorUtils(final ActorSystem actorSystem, final ActorRef shardManager,
final ClusterWrapper clusterWrapper, final Configuration configuration,
final DatastoreContext datastoreContext, final PrimaryShardInfoFutureCache primaryShardInfoCache) {
this.actorSystem = actorSystem;
final ClusterWrapper clusterWrapper, final Configuration configuration,
final DatastoreContext datastoreContext, final PrimaryShardInfoFutureCache primaryShardInfoCache) {
this.actorSystem = actorSystem;
public class TransactionRateLimiter {
private static final Logger LOG = LoggerFactory.getLogger(TransactionRateLimiter.class);
public class TransactionRateLimiter {
private static final Logger LOG = LoggerFactory.getLogger(TransactionRateLimiter.class);
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final long commitTimeoutInSeconds;
private final String dataStoreName;
private final RateLimiter txRateLimiter;
private final long commitTimeoutInSeconds;
private final String dataStoreName;
private final RateLimiter txRateLimiter;
private volatile long pollOnCount = 1;
private volatile long pollOnCount = 1;
- public TransactionRateLimiter(ActorContext actorContext) {
- this.actorContext = actorContext;
- this.commitTimeoutInSeconds = actorContext.getDatastoreContext().getShardTransactionCommitTimeoutInSeconds();
- this.dataStoreName = actorContext.getDataStoreName();
- this.txRateLimiter = RateLimiter.create(actorContext.getDatastoreContext()
+ public TransactionRateLimiter(ActorUtils actorUtils) {
+ this.actorUtils = actorUtils;
+ this.commitTimeoutInSeconds = actorUtils.getDatastoreContext().getShardTransactionCommitTimeoutInSeconds();
+ this.dataStoreName = actorUtils.getDataStoreName();
+ this.txRateLimiter = RateLimiter.create(actorUtils.getDatastoreContext()
.getTransactionCreationInitialRateLimit());
}
.getTransactionCreationInitialRateLimit());
}
private void adjustRateLimit() {
final long count = acquireCount.incrementAndGet();
if (count >= pollOnCount) {
private void adjustRateLimit() {
final long count = acquireCount.incrementAndGet();
if (count >= pollOnCount) {
- final Timer commitTimer = actorContext.getOperationTimer(ActorContext.COMMIT);
+ final Timer commitTimer = actorUtils.getOperationTimer(ActorUtils.COMMIT);
double newRateLimit = calculateNewRateLimit(commitTimer, commitTimeoutInSeconds);
if (newRateLimit < 1.0) {
double newRateLimit = calculateNewRateLimit(commitTimer, commitTimeoutInSeconds);
if (newRateLimit < 1.0) {
- double newRateLimit = calculateNewRateLimit(actorContext.getOperationTimer(name, ActorContext.COMMIT),
+ double newRateLimit = calculateNewRateLimit(actorUtils.getOperationTimer(name, ActorUtils.COMMIT),
this.commitTimeoutInSeconds);
if (newRateLimit > 0.0) {
LOG.debug("On unused Tx - data Store {} commit rateLimit adjusted to {}",
this.commitTimeoutInSeconds);
if (newRateLimit > 0.0) {
LOG.debug("On unused Tx - data Store {} commit rateLimit adjusted to {}",
import java.util.concurrent.ConcurrentHashMap;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
import java.util.concurrent.ConcurrentHashMap;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
import org.opendaylight.controller.cluster.dom.api.LeaderLocation;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
import org.opendaylight.controller.cluster.dom.api.LeaderLocation;
private final Collection<LeaderLocationListener> listeners = ConcurrentHashMap.newKeySet();
private final DOMDataTreeIdentifier prefix;
private final Collection<LeaderLocationListener> listeners = ConcurrentHashMap.newKeySet();
private final DOMDataTreeIdentifier prefix;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final Timeout makeLeaderLocalTimeout;
private ActorRef roleChangeListenerActor;
private final Timeout makeLeaderLocalTimeout;
private ActorRef roleChangeListenerActor;
private volatile LeaderLocation currentLeader = LeaderLocation.UNKNOWN;
private volatile boolean closed = false;
private volatile LeaderLocation currentLeader = LeaderLocation.UNKNOWN;
private volatile boolean closed = false;
- CDSShardAccessImpl(final DOMDataTreeIdentifier prefix, final ActorContext actorContext) {
+ CDSShardAccessImpl(final DOMDataTreeIdentifier prefix, final ActorUtils actorUtils) {
this.prefix = requireNonNull(prefix);
this.prefix = requireNonNull(prefix);
- this.actorContext = requireNonNull(actorContext);
+ this.actorUtils = requireNonNull(actorUtils);
this.makeLeaderLocalTimeout =
this.makeLeaderLocalTimeout =
- new Timeout(actorContext.getDatastoreContext().getShardLeaderElectionTimeout().duration().$times(2));
+ new Timeout(actorUtils.getDatastoreContext().getShardLeaderElectionTimeout().duration().$times(2));
// register RoleChangeListenerActor
// TODO Maybe we should do this in async
final Optional<ActorRef> localShardReply =
// register RoleChangeListenerActor
// TODO Maybe we should do this in async
final Optional<ActorRef> localShardReply =
- actorContext.findLocalShard(ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
+ actorUtils.findLocalShard(ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
checkState(localShardReply.isPresent(),
"Local shard for {} not present. Cannot register RoleChangeListenerActor", prefix);
roleChangeListenerActor =
checkState(localShardReply.isPresent(),
"Local shard for {} not present. Cannot register RoleChangeListenerActor", prefix);
roleChangeListenerActor =
- actorContext.getActorSystem().actorOf(RoleChangeListenerActor.props(localShardReply.get(), this));
+ actorUtils.getActorSystem().actorOf(RoleChangeListenerActor.props(localShardReply.get(), this));
}
private void checkNotClosed() {
}
private void checkNotClosed() {
// TODO can we cache local shard actorRef?
final Future<ActorRef> localShardReply =
// TODO can we cache local shard actorRef?
final Future<ActorRef> localShardReply =
- actorContext.findLocalShardAsync(ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
+ actorUtils.findLocalShardAsync(ClusterUtils.getCleanShardName(prefix.getRootIdentifier()));
// we have to tell local shard to make leader local
final scala.concurrent.Promise<Object> makeLeaderLocalAsk = Futures.promise();
// we have to tell local shard to make leader local
final scala.concurrent.Promise<Object> makeLeaderLocalAsk = Futures.promise();
makeLeaderLocalAsk.failure(failure);
} else {
makeLeaderLocalAsk
makeLeaderLocalAsk.failure(failure);
} else {
makeLeaderLocalAsk
- .completeWith(actorContext
+ .completeWith(actorUtils
.executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
}
}
.executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
// we have to transform make leader local request result
Future<Void> makeLeaderLocalFuture = makeLeaderLocalAsk.future()
// we have to transform make leader local request result
Future<Void> makeLeaderLocalFuture = makeLeaderLocalAsk.future()
// wrap exception in LeadershipTransferFailedEx
return new LeadershipTransferFailedException("Leadership transfer failed", parameter);
}
// wrap exception in LeadershipTransferFailedEx
return new LeadershipTransferFailedException("Leadership transfer failed", parameter);
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
return FutureConverters.toJava(makeLeaderLocalFuture);
}
return FutureConverters.toJava(makeLeaderLocalFuture);
}
.build());
// XXX: can we guarantee that the root is present in the schemacontext?
.build());
// XXX: can we guarantee that the root is present in the schemacontext?
- this.dataTree.setSchemaContext(distributedDataStore.getActorContext().getSchemaContext());
+ this.dataTree.setSchemaContext(distributedDataStore.getActorUtils().getSchemaContext());
this.shardPath = prefix.getRootIdentifier();
this.childShards = childShards;
}
this.shardPath = prefix.getRootIdentifier();
this.childShards = childShards;
}
package org.opendaylight.controller.cluster.sharding;
import static akka.actor.ActorRef.noSender;
package org.opendaylight.controller.cluster.sharding;
import static akka.actor.ActorRef.noSender;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ModuleShardStrategy;
import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ModuleShardStrategy;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
new ShardedDataTreeActorCreator()
.setShardingService(this)
.setActorSystem(actorSystem)
new ShardedDataTreeActorCreator()
.setShardingService(this)
.setActorSystem(actorSystem)
- .setClusterWrapper(distributedConfigDatastore.getActorContext().getClusterWrapper())
+ .setClusterWrapper(distributedConfigDatastore.getActorUtils().getClusterWrapper())
.setDistributedConfigDatastore(distributedConfigDatastore)
.setDistributedOperDatastore(distributedOperDatastore)
.setLookupTaskMaxRetries(LOOKUP_TASK_MAX_RETRIES),
ACTOR_ID);
.setDistributedConfigDatastore(distributedConfigDatastore)
.setDistributedOperDatastore(distributedOperDatastore)
.setLookupTaskMaxRetries(LOOKUP_TASK_MAX_RETRIES),
ACTOR_ID);
- this.memberName = distributedConfigDatastore.getActorContext().getCurrentMemberName();
+ this.memberName = distributedConfigDatastore.getActorUtils().getCurrentMemberName();
updateHandler = new PrefixedShardConfigUpdateHandler(shardedDataTreeActor,
updateHandler = new PrefixedShardConfigUpdateHandler(shardedDataTreeActor,
- distributedConfigDatastore.getActorContext().getCurrentMemberName());
+ distributedConfigDatastore.getActorUtils().getCurrentMemberName());
LOG.debug("{} - Starting prefix configuration shards", memberName);
createPrefixConfigShard(distributedConfigDatastore);
LOG.debug("{} - Starting prefix configuration shards", memberName);
createPrefixConfigShard(distributedConfigDatastore);
}
private static void createPrefixConfigShard(final AbstractDataStore dataStore) {
}
private static void createPrefixConfigShard(final AbstractDataStore dataStore) {
- Configuration configuration = dataStore.getActorContext().getConfiguration();
+ Configuration configuration = dataStore.getActorUtils().getConfiguration();
Collection<MemberName> memberNames = configuration.getUniqueMemberNamesForAllShards();
CreateShard createShardMessage =
new CreateShard(new ModuleShardConfiguration(PrefixShards.QNAME.getNamespace(),
"prefix-shard-configuration", ClusterUtils.PREFIX_CONFIG_SHARD_ID, ModuleShardStrategy.NAME,
memberNames),
Collection<MemberName> memberNames = configuration.getUniqueMemberNamesForAllShards();
CreateShard createShardMessage =
new CreateShard(new ModuleShardConfiguration(PrefixShards.QNAME.getNamespace(),
"prefix-shard-configuration", ClusterUtils.PREFIX_CONFIG_SHARD_ID, ModuleShardStrategy.NAME,
memberNames),
- Shard.builder(), dataStore.getActorContext().getDatastoreContext());
+ Shard.builder(), dataStore.getActorUtils().getDatastoreContext());
- dataStore.getActorContext().getShardManager().tell(createShardMessage, noSender());
+ dataStore.getActorUtils().getShardManager().tell(createShardMessage, noSender());
LOG.debug("{}: Prefix configuration shards ready - creating clients", memberName);
configurationShardMap.put(LogicalDatastoreType.CONFIGURATION,
createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
LOG.debug("{}: Prefix configuration shards ready - creating clients", memberName);
configurationShardMap.put(LogicalDatastoreType.CONFIGURATION,
createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
- distributedConfigDatastore.getActorContext()));
+ distributedConfigDatastore.getActorUtils()));
} catch (final DOMDataTreeShardCreationFailedException e) {
throw new IllegalStateException(
"Unable to create datastoreClient for config DS prefix configuration shard.", e);
} catch (final DOMDataTreeShardCreationFailedException e) {
throw new IllegalStateException(
"Unable to create datastoreClient for config DS prefix configuration shard.", e);
try {
configurationShardMap.put(LogicalDatastoreType.OPERATIONAL,
createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
try {
configurationShardMap.put(LogicalDatastoreType.OPERATIONAL,
createDatastoreClient(ClusterUtils.PREFIX_CONFIG_SHARD_ID,
- distributedOperDatastore.getActorContext()));
+ distributedOperDatastore.getActorUtils()));
} catch (final DOMDataTreeShardCreationFailedException e) {
throw new IllegalStateException(
} catch (final DOMDataTreeShardCreationFailedException e) {
throw new IllegalStateException(
updateHandler.initListener(distributedConfigDatastore, LogicalDatastoreType.CONFIGURATION);
updateHandler.initListener(distributedOperDatastore, LogicalDatastoreType.OPERATIONAL);
updateHandler.initListener(distributedConfigDatastore, LogicalDatastoreType.CONFIGURATION);
updateHandler.initListener(distributedOperDatastore, LogicalDatastoreType.OPERATIONAL);
- distributedConfigDatastore.getActorContext().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
- distributedOperDatastore.getActorContext().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
+ distributedConfigDatastore.getActorUtils().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
+ distributedOperDatastore.getActorUtils().getShardManager().tell(InitConfigListener.INSTANCE, noSender());
//create shard registration for DEFAULT_SHARD
//create shard registration for DEFAULT_SHARD
LOG.debug("{} - Creating producer for {}", memberName, subtrees);
final DOMDataTreeProducer producer = shardedDOMDataTree.createProducer(subtrees);
LOG.debug("{} - Creating producer for {}", memberName, subtrees);
final DOMDataTreeProducer producer = shardedDOMDataTree.createProducer(subtrees);
- final Object response = distributedConfigDatastore.getActorContext()
+ final Object response = distributedConfigDatastore.getActorUtils()
.executeOperation(shardedDataTreeActor, new ProducerCreated(subtrees));
if (response == null) {
LOG.debug("{} - Received success from remote nodes, creating producer:{}", memberName, subtrees);
return new ProxyProducer(producer, subtrees, shardedDataTreeActor,
.executeOperation(shardedDataTreeActor, new ProducerCreated(subtrees));
if (response == null) {
LOG.debug("{} - Received success from remote nodes, creating producer:{}", memberName, subtrees);
return new ProxyProducer(producer, subtrees, shardedDataTreeActor,
- distributedConfigDatastore.getActorContext(), shards);
+ distributedConfigDatastore.getActorUtils(), shards);
}
closeProducer(producer);
}
closeProducer(producer);
try (DOMDataTreeProducer producer = localCreateProducer(Collections.singletonList(prefix))) {
final Entry<DataStoreClient, ActorRef> entry =
try (DOMDataTreeProducer producer = localCreateProducer(Collections.singletonList(prefix))) {
final Entry<DataStoreClient, ActorRef> entry =
- createDatastoreClient(shardName, distributedDataStore.getActorContext());
+ createDatastoreClient(shardName, distributedDataStore.getActorUtils());
final DistributedShardFrontend shard =
new DistributedShardFrontend(distributedDataStore, entry.getKey(), prefix);
final DistributedShardFrontend shard =
new DistributedShardFrontend(distributedDataStore, entry.getKey(), prefix);
} catch (final DOMDataTreeShardingConflictException e) {
LOG.error("{}: Prefix {} is already occupied by another shard",
} catch (final DOMDataTreeShardingConflictException e) {
LOG.error("{}: Prefix {} is already occupied by another shard",
- distributedConfigDatastore.getActorContext().getClusterWrapper().getCurrentMemberName(), prefix, e);
+ distributedConfigDatastore.getActorUtils().getClusterWrapper().getCurrentMemberName(), prefix, e);
} catch (DOMDataTreeProducerException e) {
LOG.error("Unable to close producer", e);
} catch (DOMDataTreeShardCreationFailedException e) {
} catch (DOMDataTreeProducerException e) {
LOG.error("Unable to close producer", e);
} catch (DOMDataTreeShardCreationFailedException e) {
}
@SuppressWarnings("checkstyle:IllegalCatch")
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private Entry<DataStoreClient, ActorRef> createDatastoreClient(
- final String shardName, final ActorContext actorContext)
+ private Entry<DataStoreClient, ActorRef> createDatastoreClient(final String shardName, final ActorUtils actorUtils)
throws DOMDataTreeShardCreationFailedException {
LOG.debug("{}: Creating distributed datastore client for shard {}", memberName, shardName);
final Props distributedDataStoreClientProps =
throws DOMDataTreeShardCreationFailedException {
LOG.debug("{}: Creating distributed datastore client for shard {}", memberName, shardName);
final Props distributedDataStoreClientProps =
- SimpleDataStoreClientActor.props(memberName, "Shard-" + shardName, actorContext, shardName);
+ SimpleDataStoreClientActor.props(memberName, "Shard-" + shardName, actorUtils, shardName);
final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
try {
final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
try {
// TODO we don't have to do it for config and operational default shard separately. Just one of them
// should be enough
// TODO we don't have to do it for config and operational default shard separately. Just one of them
// should be enough
- final ActorContext actorContext = logicalDatastoreType == LogicalDatastoreType.CONFIGURATION
- ? distributedConfigDatastore.getActorContext() : distributedOperDatastore.getActorContext();
+ final ActorUtils actorUtils = logicalDatastoreType == LogicalDatastoreType.CONFIGURATION
+ ? distributedConfigDatastore.getActorUtils() : distributedOperDatastore.getActorUtils();
final Optional<ActorRef> defaultLocalShardOptional =
final Optional<ActorRef> defaultLocalShardOptional =
- actorContext.findLocalShard(ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
+ actorUtils.findLocalShard(ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
if (defaultLocalShardOptional.isPresent()) {
LOG.debug("{}: Default shard for {} is already started, creating just frontend", memberName,
if (defaultLocalShardOptional.isPresent()) {
LOG.debug("{}: Default shard for {} is already started, creating just frontend", memberName,
// the default shard as a prefix shard is problematic in this scenario so it is commented out. Since
// the default shard is a module-based shard by default, it makes sense to always treat it as such,
// ie bootstrap it in the same manner as the special prefix-configuration and EOS shards.
// the default shard as a prefix shard is problematic in this scenario so it is commented out. Since
// the default shard is a module-based shard by default, it makes sense to always treat it as such,
// ie bootstrap it in the same manner as the special prefix-configuration and EOS shards.
-// final Collection<MemberName> names = distributedConfigDatastore.getActorContext().getConfiguration()
+// final Collection<MemberName> names = distributedConfigDatastore.getActorUtils().getConfiguration()
// .getUniqueMemberNamesForAllShards();
// Await.result(FutureConverters.toScala(createDistributedShard(
// new DOMDataTreeIdentifier(logicalDatastoreType, YangInstanceIdentifier.EMPTY), names)),
// .getUniqueMemberNamesForAllShards();
// Await.result(FutureConverters.toScala(createDistributedShard(
// new DOMDataTreeIdentifier(logicalDatastoreType, YangInstanceIdentifier.EMPTY), names)),
private final DOMDataTreeProducer delegate;
private final Collection<DOMDataTreeIdentifier> subtrees;
private final ActorRef shardDataTreeActor;
private final DOMDataTreeProducer delegate;
private final Collection<DOMDataTreeIdentifier> subtrees;
private final ActorRef shardDataTreeActor;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
@GuardedBy("shardAccessMap")
private final Map<DOMDataTreeIdentifier, CDSShardAccessImpl> shardAccessMap = new HashMap<>();
@GuardedBy("shardAccessMap")
private final Map<DOMDataTreeIdentifier, CDSShardAccessImpl> shardAccessMap = new HashMap<>();
ProxyProducer(final DOMDataTreeProducer delegate,
final Collection<DOMDataTreeIdentifier> subtrees,
final ActorRef shardDataTreeActor,
ProxyProducer(final DOMDataTreeProducer delegate,
final Collection<DOMDataTreeIdentifier> subtrees,
final ActorRef shardDataTreeActor,
- final ActorContext actorContext,
+ final ActorUtils actorUtils,
final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shardLayout) {
final DOMDataTreePrefixTable<DOMDataTreeShardRegistration<DOMDataTreeShard>> shardLayout) {
- this.delegate = Preconditions.checkNotNull(delegate);
- this.subtrees = Preconditions.checkNotNull(subtrees);
- this.shardDataTreeActor = Preconditions.checkNotNull(shardDataTreeActor);
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.shardTable = Preconditions.checkNotNull(shardLayout);
+ this.delegate = requireNonNull(delegate);
+ this.subtrees = requireNonNull(subtrees);
+ this.shardDataTreeActor = requireNonNull(shardDataTreeActor);
+ this.actorUtils = requireNonNull(actorUtils);
+ this.shardTable = requireNonNull(shardLayout);
shardAccessMap.values().forEach(CDSShardAccessImpl::close);
}
shardAccessMap.values().forEach(CDSShardAccessImpl::close);
}
- final Object o = actorContext.executeOperation(shardDataTreeActor, new ProducerRemoved(subtrees));
+ final Object o = actorUtils.executeOperation(shardDataTreeActor, new ProducerRemoved(subtrees));
if (o instanceof DOMDataTreeProducerException) {
throw (DOMDataTreeProducerException) o;
} else if (o instanceof Throwable) {
if (o instanceof DOMDataTreeProducerException) {
throw (DOMDataTreeProducerException) o;
} else if (o instanceof Throwable) {
@Nonnull
@Override
public CDSShardAccess getShardAccess(@Nonnull final DOMDataTreeIdentifier subtree) {
@Nonnull
@Override
public CDSShardAccess getShardAccess(@Nonnull final DOMDataTreeIdentifier subtree) {
- Preconditions.checkArgument(
- subtrees.stream().anyMatch(dataTreeIdentifier -> dataTreeIdentifier.contains(subtree)),
- "Subtree %s is not controlled by this producer %s", subtree, this);
+ checkArgument(subtrees.stream().anyMatch(dataTreeIdentifier -> dataTreeIdentifier.contains(subtree)),
+ "Subtree %s is not controlled by this producer %s", subtree, this);
final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup =
shardTable.lookup(subtree);
final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup =
shardTable.lookup(subtree);
- Preconditions.checkState(lookup != null, "Subtree %s is not contained in any registered shard.", subtree);
+ checkState(lookup != null, "Subtree %s is not contained in any registered shard.", subtree);
final DOMDataTreeIdentifier lookupId = lookup.getValue().getPrefix();
final DOMDataTreeIdentifier lookupId = lookup.getValue().getPrefix();
// for same subtrees. But maybe it is not needed since there can be only one
// producer attached to some subtree at a time. And also how we can close ShardAccess
// then
// for same subtrees. But maybe it is not needed since there can be only one
// producer attached to some subtree at a time. And also how we can close ShardAccess
// then
- final CDSShardAccessImpl shardAccess = new CDSShardAccessImpl(lookupId, actorContext);
+ final CDSShardAccessImpl shardAccess = new CDSShardAccessImpl(lookupId, actorUtils);
shardAccessMap.put(lookupId, shardAccess);
return shardAccess;
}
shardAccessMap.put(lookupId, shardAccess);
return shardAccess;
}
import org.opendaylight.controller.cluster.datastore.AbstractDataStore;
import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
import org.opendaylight.controller.cluster.datastore.config.PrefixShardConfiguration;
import org.opendaylight.controller.cluster.datastore.AbstractDataStore;
import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
import org.opendaylight.controller.cluster.datastore.config.PrefixShardConfiguration;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
private final ClusterWrapper clusterWrapper;
// helper actorContext used only for static calls to executeAsync etc
// for calls that need specific actor context tied to a datastore use the one provided in the DistributedDataStore
private final ClusterWrapper clusterWrapper;
// helper actorContext used only for static calls to executeAsync etc
// for calls that need specific actor context tied to a datastore use the one provided in the DistributedDataStore
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final ShardingServiceAddressResolver resolver;
private final AbstractDataStore distributedConfigDatastore;
private final AbstractDataStore distributedOperDatastore;
private final ShardingServiceAddressResolver resolver;
private final AbstractDataStore distributedConfigDatastore;
private final AbstractDataStore distributedOperDatastore;
distributedConfigDatastore = builder.getDistributedConfigDatastore();
distributedOperDatastore = builder.getDistributedOperDatastore();
lookupTaskMaxRetries = builder.getLookupTaskMaxRetries();
distributedConfigDatastore = builder.getDistributedConfigDatastore();
distributedOperDatastore = builder.getDistributedOperDatastore();
lookupTaskMaxRetries = builder.getLookupTaskMaxRetries();
- actorContext = distributedConfigDatastore.getActorContext();
+ actorUtils = distributedConfigDatastore.getActorUtils();
resolver = new ShardingServiceAddressResolver(
DistributedShardedDOMDataTree.ACTOR_ID, clusterWrapper.getCurrentMemberName());
resolver = new ShardingServiceAddressResolver(
DistributedShardedDOMDataTree.ACTOR_ID, clusterWrapper.getCurrentMemberName());
final ActorSelection actorSelection = actorSystem.actorSelection(address);
futures.add(
FutureConverters.toJava(
final ActorSelection actorSelection = actorSystem.actorSelection(address);
futures.add(
FutureConverters.toJava(
- actorContext.executeOperationAsync(
+ actorUtils.executeOperationAsync(
actorSelection, new NotifyProducerCreated(subtrees), DEFAULT_ASK_TIMEOUT))
.toCompletableFuture());
}
actorSelection, new NotifyProducerCreated(subtrees), DEFAULT_ASK_TIMEOUT))
.toCompletableFuture());
}
final ActorSelection selection = actorSystem.actorSelection(address);
futures.add(FutureConverters.toJava(
final ActorSelection selection = actorSystem.actorSelection(address);
futures.add(FutureConverters.toJava(
- actorContext.executeOperationAsync(selection, new NotifyProducerRemoved(message.getSubtrees())))
+ actorUtils.executeOperationAsync(selection, new NotifyProducerRemoved(message.getSubtrees())))
.toCompletableFuture());
}
.toCompletableFuture());
}
final DOMDataTreeIdentifier prefix = message.getPrefix();
final DOMDataTreeIdentifier prefix = message.getPrefix();
- final ActorContext context = prefix.getDatastoreType() == LogicalDatastoreType.CONFIGURATION
- ? distributedConfigDatastore.getActorContext() : distributedOperDatastore.getActorContext();
+ final ActorUtils context = prefix.getDatastoreType() == LogicalDatastoreType.CONFIGURATION
+ ? distributedConfigDatastore.getActorUtils() : distributedOperDatastore.getActorUtils();
// schedule a notification task for the reply
actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL,
// schedule a notification task for the reply
actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL,
final ShardRemovalLookupTask removalTask =
new ShardRemovalLookupTask(actorSystem, getSender(),
final ShardRemovalLookupTask removalTask =
new ShardRemovalLookupTask(actorSystem, getSender(),
- actorContext, message.getPrefix(), lookupTaskMaxRetries);
+ actorUtils, message.getPrefix(), lookupTaskMaxRetries);
actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL, removalTask, actorSystem.dispatcher());
}
actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL, removalTask, actorSystem.dispatcher());
}
private void onStartConfigShardLookup(final StartConfigShardLookup message) {
LOG.debug("Received StartConfigShardLookup: {}", message);
private void onStartConfigShardLookup(final StartConfigShardLookup message) {
LOG.debug("Received StartConfigShardLookup: {}", message);
- final ActorContext context =
+ final ActorUtils context =
message.getType().equals(LogicalDatastoreType.CONFIGURATION)
message.getType().equals(LogicalDatastoreType.CONFIGURATION)
- ? distributedConfigDatastore.getActorContext() : distributedOperDatastore.getActorContext();
+ ? distributedConfigDatastore.getActorUtils() : distributedOperDatastore.getActorUtils();
// schedule a notification task for the reply
actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL,
// schedule a notification task for the reply
actorSystem.scheduler().scheduleOnce(SHARD_LOOKUP_TASK_INTERVAL,
private final ActorSystem system;
private final ActorRef replyTo;
private final ClusterWrapper clusterWrapper;
private final ActorSystem system;
private final ActorRef replyTo;
private final ClusterWrapper clusterWrapper;
- private final ActorContext context;
+ private final ActorUtils context;
private final DistributedShardedDOMDataTree shardingService;
private final DOMDataTreeIdentifier toLookup;
private final int lookupMaxRetries;
private final DistributedShardedDOMDataTree shardingService;
private final DOMDataTreeIdentifier toLookup;
private final int lookupMaxRetries;
ShardCreationLookupTask(final ActorSystem system,
final ActorRef replyTo,
final ClusterWrapper clusterWrapper,
ShardCreationLookupTask(final ActorSystem system,
final ActorRef replyTo,
final ClusterWrapper clusterWrapper,
- final ActorContext context,
+ final ActorUtils context,
final DistributedShardedDOMDataTree shardingService,
final DOMDataTreeIdentifier toLookup,
final int lookupMaxRetries) {
final DistributedShardedDOMDataTree shardingService,
final DOMDataTreeIdentifier toLookup,
final int lookupMaxRetries) {
private final ActorSystem system;
private final ActorRef replyTo;
private final ActorSystem system;
private final ActorRef replyTo;
- private final ActorContext context;
+ private final ActorUtils context;
private final ClusterWrapper clusterWrapper;
private final ActorRef shard;
private final DistributedShardedDOMDataTree shardingService;
private final ClusterWrapper clusterWrapper;
private final ActorRef shard;
private final DistributedShardedDOMDataTree shardingService;
ShardLeaderLookupTask(final ActorSystem system,
final ActorRef replyTo,
ShardLeaderLookupTask(final ActorSystem system,
final ActorRef replyTo,
- final ActorContext context,
+ final ActorUtils context,
final ClusterWrapper clusterWrapper,
final ActorRef shard,
final DistributedShardedDOMDataTree shardingService,
final ClusterWrapper clusterWrapper,
final ActorRef shard,
final DistributedShardedDOMDataTree shardingService,
private final ActorSystem system;
private final ActorRef replyTo;
private final ActorSystem system;
private final ActorRef replyTo;
- private final ActorContext context;
+ private final ActorUtils context;
private final DOMDataTreeIdentifier toLookup;
ShardRemovalLookupTask(final ActorSystem system,
final ActorRef replyTo,
private final DOMDataTreeIdentifier toLookup;
ShardRemovalLookupTask(final ActorSystem system,
final ActorRef replyTo,
- final ActorContext context,
+ final ActorUtils context,
final DOMDataTreeIdentifier toLookup,
final int lookupMaxRetries) {
super(replyTo, lookupMaxRetries);
final DOMDataTreeIdentifier toLookup,
final int lookupMaxRetries) {
super(replyTo, lookupMaxRetries);
private final ActorSystem system;
private final ActorRef replyTo;
private final ActorSystem system;
private final ActorRef replyTo;
- private final ActorContext context;
+ private final ActorUtils context;
ConfigShardLookupTask(final ActorSystem system,
final ActorRef replyTo,
ConfigShardLookupTask(final ActorSystem system,
final ActorRef replyTo,
- final ActorContext context,
+ final ActorUtils context,
final StartConfigShardLookup message,
final int lookupMaxRetries) {
super(replyTo, lookupMaxRetries);
final StartConfigShardLookup message,
final int lookupMaxRetries) {
super(replyTo, lookupMaxRetries);
private final ActorSystem system;
private final ActorRef replyTo;
private final ActorSystem system;
private final ActorRef replyTo;
- private final ActorContext context;
+ private final ActorUtils context;
private final ClusterWrapper clusterWrapper;
private final ActorRef shard;
ConfigShardReadinessTask(final ActorSystem system,
final ActorRef replyTo,
private final ClusterWrapper clusterWrapper;
private final ActorRef shard;
ConfigShardReadinessTask(final ActorSystem system,
final ActorRef replyTo,
- final ActorContext context,
+ final ActorUtils context,
final ClusterWrapper clusterWrapper,
final ActorRef shard,
final int lookupMaxRetries) {
final ClusterWrapper clusterWrapper,
final ActorRef shard,
final int lookupMaxRetries) {
<bean id="distributedEntityOwnershipService" class="org.opendaylight.controller.cluster.datastore.entityownership.DistributedEntityOwnershipService"
factory-method="start" destroy-method="close">
<argument>
<bean id="distributedEntityOwnershipService" class="org.opendaylight.controller.cluster.datastore.entityownership.DistributedEntityOwnershipService"
factory-method="start" destroy-method="close">
<argument>
- <bean factory-ref="operDatastore" factory-method="getActorContext"/>
+ <bean factory-ref="operDatastore" factory-method="getActorUtils"/>
</argument>
<argument ref="selectionStrategyConfig"/>
</bean>
</argument>
<argument ref="selectionStrategyConfig"/>
</bean>
import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
private DataStoreClient clientActor;
@Mock
private DataStoreClient clientActor;
@Mock
- private ActorContext actorContext;
+ private ActorUtils actorUtils;
@Mock
private ClientLocalHistory clientLocalHistory;
@Mock
private ClientLocalHistory clientLocalHistory;
public void setUp() {
MockitoAnnotations.initMocks(this);
public void setUp() {
MockitoAnnotations.initMocks(this);
- when(actorContext.getSchemaContext()).thenReturn(SCHEMA_CONTEXT);
- when(actorContext.getDatastoreContext()).thenReturn(DatastoreContext.newBuilder().build());
+ when(actorUtils.getSchemaContext()).thenReturn(SCHEMA_CONTEXT);
+ when(actorUtils.getDatastoreContext()).thenReturn(DatastoreContext.newBuilder().build());
when(clientTransaction.getIdentifier()).thenReturn(TRANSACTION_IDENTIFIER);
when(clientSnapshot.getIdentifier()).thenReturn(TRANSACTION_IDENTIFIER);
when(clientTransaction.getIdentifier()).thenReturn(TRANSACTION_IDENTIFIER);
when(clientSnapshot.getIdentifier()).thenReturn(TRANSACTION_IDENTIFIER);
@Test
public void testCreateTransactionChain() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
@Test
public void testCreateTransactionChain() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
- actorContext, UNKNOWN_ID, clientActor)) {
+ actorUtils, UNKNOWN_ID, clientActor)) {
final DOMStoreTransactionChain txChain = clientBackedDataStore.createTransactionChain();
assertNotNull(txChain);
verify(clientActor, Mockito.times(1)).createLocalHistory();
final DOMStoreTransactionChain txChain = clientBackedDataStore.createTransactionChain();
assertNotNull(txChain);
verify(clientActor, Mockito.times(1)).createLocalHistory();
@Test
public void testNewReadOnlyTransaction() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
@Test
public void testNewReadOnlyTransaction() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
- actorContext, UNKNOWN_ID, clientActor)) {
+ actorUtils, UNKNOWN_ID, clientActor)) {
final DOMStoreReadTransaction tx = clientBackedDataStore.newReadOnlyTransaction();
assertNotNull(tx);
verify(clientActor, Mockito.times(1)).createSnapshot();
final DOMStoreReadTransaction tx = clientBackedDataStore.newReadOnlyTransaction();
assertNotNull(tx);
verify(clientActor, Mockito.times(1)).createSnapshot();
@Test
public void testNewWriteOnlyTransaction() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
@Test
public void testNewWriteOnlyTransaction() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
- actorContext, UNKNOWN_ID, clientActor)) {
+ actorUtils, UNKNOWN_ID, clientActor)) {
final DOMStoreWriteTransaction tx = clientBackedDataStore.newWriteOnlyTransaction();
assertNotNull(tx);
verify(clientActor, Mockito.times(1)).createTransaction();
final DOMStoreWriteTransaction tx = clientBackedDataStore.newWriteOnlyTransaction();
assertNotNull(tx);
verify(clientActor, Mockito.times(1)).createTransaction();
@Test
public void testNewReadWriteTransaction() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
@Test
public void testNewReadWriteTransaction() {
try (ClientBackedDataStore clientBackedDataStore = new ClientBackedDataStore(
- actorContext, UNKNOWN_ID, clientActor)) {
+ actorUtils, UNKNOWN_ID, clientActor)) {
final DOMStoreReadWriteTransaction tx = clientBackedDataStore.newReadWriteTransaction();
assertNotNull(tx);
verify(clientActor, Mockito.times(1)).createTransaction();
final DOMStoreReadWriteTransaction tx = clientBackedDataStore.newReadWriteTransaction();
assertNotNull(tx);
verify(clientActor, Mockito.times(1)).createTransaction();
import org.opendaylight.controller.cluster.access.concepts.Response;
import org.opendaylight.controller.cluster.access.concepts.SuccessEnvelope;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.access.concepts.Response;
import org.opendaylight.controller.cluster.access.concepts.SuccessEnvelope;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
final TestProbe clientContextProbe = new TestProbe(system, "client-context");
backendProbe = new TestProbe(system, "backend");
//create handle dependencies
final TestProbe clientContextProbe = new TestProbe(system, "client-context");
backendProbe = new TestProbe(system, "backend");
//create handle dependencies
- final ActorContext actorContext = createActorContextMock(system, contextProbe.ref());
+ final ActorUtils actorUtils = createActorContextMock(system, contextProbe.ref());
final ClientActorContext clientContext =
AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
final ClientActorContext clientContext =
AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
- client = new SimpleDataStoreClientBehavior(clientContext, actorContext, "shard");
+ client = new SimpleDataStoreClientBehavior(clientContext, actorUtils, "shard");
client.createLocalHistory();
parent = new SingleClientHistory(client, HISTORY_ID);
//connect client
client.createLocalHistory();
parent = new SingleClientHistory(client, HISTORY_ID);
//connect client
return dataTreeSnapshot;
}
return dataTreeSnapshot;
}
- private static ActorContext createActorContextMock(final ActorSystem system, final ActorRef actor) {
- final ActorContext mock = mock(ActorContext.class);
+ private static ActorUtils createActorContextMock(final ActorSystem system, final ActorRef actor) {
+ final ActorUtils mock = mock(ActorUtils.class);
final Promise<PrimaryShardInfo> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
final Promise<PrimaryShardInfo> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
import org.opendaylight.controller.cluster.access.client.ConnectedClientConnection;
import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.access.client.ConnectedClientConnection;
import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import scala.concurrent.Promise;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import scala.concurrent.Promise;
Assert.assertNull(reconnectCohort);
}
Assert.assertNull(reconnectCohort);
}
- protected static ActorContext createActorContextMock(final ActorSystem system, final ActorRef actor) {
- final ActorContext mock = mock(ActorContext.class);
+ protected static ActorUtils createActorUtilsMock(final ActorSystem system, final ActorRef actor) {
+ final ActorUtils mock = mock(ActorUtils.class);
final Promise<PrimaryShardInfo> promise = new DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
final Promise<PrimaryShardInfo> promise = new DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.access.commands.ConnectClientRequest;
import org.opendaylight.controller.cluster.access.commands.ConnectClientSuccess;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeModification;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.tree.CursorAwareDataTreeModification;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
system = ActorSystem.apply();
clientActorProbe = new TestProbe(system, "client");
actorContextProbe = new TestProbe(system, "actor-context");
system = ActorSystem.apply();
clientActorProbe = new TestProbe(system, "client");
actorContextProbe = new TestProbe(system, "actor-context");
- final ActorContext context = createActorContextMock(system, actorContextProbe.ref());
+ final ActorUtils context = createActorContextMock(system, actorContextProbe.ref());
clientContext =
AccessClientUtil.createClientActorContext(system, clientActorProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
behavior = createBehavior(clientContext, context);
clientContext =
AccessClientUtil.createClientActorContext(system, clientActorProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
behavior = createBehavior(clientContext, context);
@SuppressWarnings("checkstyle:hiddenField")
protected abstract AbstractDataStoreClientBehavior createBehavior(ClientActorContext clientContext,
@SuppressWarnings("checkstyle:hiddenField")
protected abstract AbstractDataStoreClientBehavior createBehavior(ClientActorContext clientContext,
@After
public void tearDown() {
@After
public void tearDown() {
verify(modification).readNode(YangInstanceIdentifier.EMPTY);
}
verify(modification).readNode(YangInstanceIdentifier.EMPTY);
}
- private static ActorContext createActorContextMock(final ActorSystem system, final ActorRef actor) {
- final ActorContext mock = mock(ActorContext.class);
+ private static ActorUtils createActorContextMock(final ActorSystem system, final ActorRef actor) {
+ final ActorUtils mock = mock(ActorUtils.class);
final Promise<PrimaryShardInfo> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
final Promise<PrimaryShardInfo> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
import org.opendaylight.controller.cluster.access.client.AccessClientUtil;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.access.client.AccessClientUtil;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
public class ClientLocalHistoryTest extends AbstractClientHistoryTest<ClientLocalHistory> {
private ActorSystem system;
public class ClientLocalHistoryTest extends AbstractClientHistoryTest<ClientLocalHistory> {
private ActorSystem system;
final TestProbe actorContextProbe = new TestProbe(system, "actor-context");
clientActorContext = AccessClientUtil.createClientActorContext(
system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
final TestProbe actorContextProbe = new TestProbe(system, "actor-context");
clientActorContext = AccessClientUtil.createClientActorContext(
system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
- final ActorContext actorContextMock = createActorContextMock(system, actorContextProbe.ref());
- behavior = new SimpleDataStoreClientBehavior(clientActorContext, actorContextMock, SHARD_NAME);
+ final ActorUtils actorUtilsMock = createActorUtilsMock(system, actorContextProbe.ref());
+ behavior = new SimpleDataStoreClientBehavior(clientActorContext, actorUtilsMock, SHARD_NAME);
object = new ClientLocalHistory(behavior, HISTORY_ID);
}
object = new ClientLocalHistory(behavior, HISTORY_ID);
}
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
public class DistributedDataStoreClientBehaviorTest extends AbstractDataStoreClientBehaviorTest {
@Override
protected AbstractDataStoreClientBehavior createBehavior(final ClientActorContext clientContext,
public class DistributedDataStoreClientBehaviorTest extends AbstractDataStoreClientBehaviorTest {
@Override
protected AbstractDataStoreClientBehavior createBehavior(final ClientActorContext clientContext,
- final ActorContext context) {
+ final ActorUtils context) {
final ShardStrategyFactory factory = mock(ShardStrategyFactory.class);
final ShardStrategy strategy = mock(ShardStrategy.class);
when(strategy.findShard(any())).thenReturn(SHARD);
final ShardStrategyFactory factory = mock(ShardStrategyFactory.class);
final ShardStrategy strategy = mock(ShardStrategy.class);
when(strategy.findShard(any())).thenReturn(SHARD);
import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
import org.opendaylight.yangtools.concepts.Registration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
import org.opendaylight.yangtools.concepts.Registration;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
shardManagerProbe = new TestProbe(system, "ShardManager");
shardManagerProbe = new TestProbe(system, "ShardManager");
- final ActorContext actorContext = createActorContextMock(system, contextProbe.ref());
- when(actorContext.getShardManager()).thenReturn(shardManagerProbe.ref());
+ final ActorUtils actorUtils = createActorUtilsMock(system, contextProbe.ref());
+ when(actorUtils.getShardManager()).thenReturn(shardManagerProbe.ref());
- moduleShardBackendResolver = new ModuleShardBackendResolver(CLIENT_ID, actorContext);
- when(actorContext.getShardStrategyFactory()).thenReturn(shardStrategyFactory);
+ moduleShardBackendResolver = new ModuleShardBackendResolver(CLIENT_ID, actorUtils);
+ when(actorUtils.getShardStrategyFactory()).thenReturn(shardStrategyFactory);
when(shardStrategyFactory.getStrategy(YangInstanceIdentifier.EMPTY)).thenReturn(shardStrategy);
final PrimaryShardInfoFutureCache cache = new PrimaryShardInfoFutureCache();
when(shardStrategyFactory.getStrategy(YangInstanceIdentifier.EMPTY)).thenReturn(shardStrategy);
final PrimaryShardInfoFutureCache cache = new PrimaryShardInfoFutureCache();
- when(actorContext.getPrimaryShardInfoCache()).thenReturn(cache);
+ when(actorUtils.getPrimaryShardInfoCache()).thenReturn(cache);
verifyNoMoreInteractions(mockCallback);
}
verifyNoMoreInteractions(mockCallback);
}
- private static ActorContext createActorContextMock(final ActorSystem system, final ActorRef actor) {
- final ActorContext mock = mock(ActorContext.class);
+ private static ActorUtils createActorUtilsMock(final ActorSystem system, final ActorRef actor) {
+ final ActorUtils mock = mock(ActorUtils.class);
final Promise<PrimaryShardInfo> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
final Promise<PrimaryShardInfo> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
final ActorSelection selection = system.actorSelection(actor.path());
final PrimaryShardInfo shardInfo = new PrimaryShardInfo(selection, (short) 0);
package org.opendaylight.controller.cluster.databroker.actors.dds;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
package org.opendaylight.controller.cluster.databroker.actors.dds;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
public class SimpleDataStoreClientBehaviorTest extends AbstractDataStoreClientBehaviorTest {
@Override
protected AbstractDataStoreClientBehavior createBehavior(final ClientActorContext clientContext,
public class SimpleDataStoreClientBehaviorTest extends AbstractDataStoreClientBehaviorTest {
@Override
protected AbstractDataStoreClientBehavior createBehavior(final ClientActorContext clientContext,
- final ActorContext context) {
+ final ActorUtils context) {
return new SimpleDataStoreClientBehavior(clientContext, context, SHARD);
}
return new SimpleDataStoreClientBehavior(clientContext, context, SHARD);
}
import org.opendaylight.controller.cluster.access.client.AccessClientUtil;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.access.client.AccessClientUtil;
import org.opendaylight.controller.cluster.access.client.ClientActorContext;
import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
public class SingleClientHistoryTest extends AbstractClientHistoryTest<SingleClientHistory> {
private ActorSystem system;
public class SingleClientHistoryTest extends AbstractClientHistoryTest<SingleClientHistory> {
private ActorSystem system;
final TestProbe actorContextProbe = new TestProbe(system, "actor-context");
clientActorContext = AccessClientUtil.createClientActorContext(
system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
final TestProbe actorContextProbe = new TestProbe(system, "actor-context");
clientActorContext = AccessClientUtil.createClientActorContext(
system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
- final ActorContext actorContextMock = createActorContextMock(system, actorContextProbe.ref());
- behavior = new SimpleDataStoreClientBehavior(clientActorContext, actorContextMock, SHARD_NAME);
+ final ActorUtils actorUtilsMock = createActorUtilsMock(system, actorContextProbe.ref());
+ behavior = new SimpleDataStoreClientBehavior(clientActorContext, actorUtilsMock, SHARD_NAME);
object = new SingleClientHistory(behavior, HISTORY_ID);
}
object = new SingleClientHistory(behavior, HISTORY_ID);
}
import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
import org.opendaylight.controller.cluster.datastore.shardstrategy.DefaultShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
- protected ActorContext mockActorContext;
+ protected ActorUtils mockActorContext;
protected TransactionContextFactory mockComponentFactory;
protected TransactionContextFactory mockComponentFactory;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
@Test(timeout = 10000)
public void testSuccessfulRegistration() {
final TestKit kit = new TestKit(getSystem());
@Test(timeout = 10000)
public void testSuccessfulRegistration() {
final TestKit kit = new TestKit(getSystem());
- ActorContext actorContext = new ActorContext(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
mock(Configuration.class));
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
mock(Configuration.class));
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
- actorContext, mockListener, path);
+ actorUtils, mockListener, path);
new Thread(() -> proxy.init("shard-1")).start();
new Thread(() -> proxy.init("shard-1")).start();
@Test(timeout = 10000)
public void testSuccessfulRegistrationForClusteredListener() {
final TestKit kit = new TestKit(getSystem());
@Test(timeout = 10000)
public void testSuccessfulRegistrationForClusteredListener() {
final TestKit kit = new TestKit(getSystem());
- ActorContext actorContext = new ActorContext(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
mock(Configuration.class));
ClusteredDOMDataTreeChangeListener mockClusteredListener = mock(
mock(Configuration.class));
ClusteredDOMDataTreeChangeListener mockClusteredListener = mock(
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<ClusteredDOMDataTreeChangeListener> proxy =
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<ClusteredDOMDataTreeChangeListener> proxy =
- new DataTreeChangeListenerProxy<>(actorContext, mockClusteredListener, path);
+ new DataTreeChangeListenerProxy<>(actorUtils, mockClusteredListener, path);
new Thread(() -> proxy.init("shard-1")).start();
new Thread(() -> proxy.init("shard-1")).start();
@Test(timeout = 10000)
public void testLocalShardNotFound() {
final TestKit kit = new TestKit(getSystem());
@Test(timeout = 10000)
public void testLocalShardNotFound() {
final TestKit kit = new TestKit(getSystem());
- ActorContext actorContext = new ActorContext(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
mock(Configuration.class));
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
mock(Configuration.class));
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
- actorContext, mockListener, path);
+ actorUtils, mockListener, path);
new Thread(() -> proxy.init("shard-1")).start();
new Thread(() -> proxy.init("shard-1")).start();
@Test(timeout = 10000)
public void testLocalShardNotInitialized() {
final TestKit kit = new TestKit(getSystem());
@Test(timeout = 10000)
public void testLocalShardNotInitialized() {
final TestKit kit = new TestKit(getSystem());
- ActorContext actorContext = new ActorContext(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
mock(Configuration.class));
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
mock(Configuration.class));
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
- actorContext, mockListener, path);
+ actorUtils, mockListener, path);
new Thread(() -> proxy.init("shard-1")).start();
new Thread(() -> proxy.init("shard-1")).start();
doReturn(mockActor).when(mockActorSystem).actorOf(any(Props.class));
ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(MoreExecutors.directExecutor());
doReturn(mockActor).when(mockActorSystem).actorOf(any(Props.class));
ExecutionContextExecutor executor = ExecutionContexts.fromExecutor(MoreExecutors.directExecutor());
- ActorContext actorContext = mock(ActorContext.class);
+ ActorUtils actorUtils = mock(ActorUtils.class);
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME);
- doReturn(executor).when(actorContext).getClientDispatcher();
- doReturn(DatastoreContext.newBuilder().build()).when(actorContext).getDatastoreContext();
- doReturn(mockActorSystem).when(actorContext).getActorSystem();
+ doReturn(executor).when(actorUtils).getClientDispatcher();
+ doReturn(DatastoreContext.newBuilder().build()).when(actorUtils).getDatastoreContext();
+ doReturn(mockActorSystem).when(actorUtils).getActorSystem();
String shardName = "shard-1";
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
String shardName = "shard-1";
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
- actorContext, mockListener, path);
+ actorUtils, mockListener, path);
- doReturn(kit.duration("5 seconds")).when(actorContext).getOperationDuration();
- doReturn(Futures.successful(kit.getRef())).when(actorContext).findLocalShardAsync(eq(shardName));
- doReturn(Futures.failed(new RuntimeException("mock"))).when(actorContext).executeOperationAsync(
+ doReturn(kit.duration("5 seconds")).when(actorUtils).getOperationDuration();
+ doReturn(Futures.successful(kit.getRef())).when(actorUtils).findLocalShardAsync(eq(shardName));
+ doReturn(Futures.failed(new RuntimeException("mock"))).when(actorUtils).executeOperationAsync(
any(ActorRef.class), any(Object.class), any(Timeout.class));
any(ActorRef.class), any(Object.class), any(Timeout.class));
- doReturn(mock(DatastoreContext.class)).when(actorContext).getDatastoreContext();
+ doReturn(mock(DatastoreContext.class)).when(actorUtils).getDatastoreContext();
@Test
public void testCloseBeforeRegistration() {
final TestKit kit = new TestKit(getSystem());
@Test
public void testCloseBeforeRegistration() {
final TestKit kit = new TestKit(getSystem());
- ActorContext actorContext = mock(ActorContext.class);
+ ActorUtils actorUtils = mock(ActorUtils.class);
String shardName = "shard-1";
String shardName = "shard-1";
- doReturn(DatastoreContext.newBuilder().build()).when(actorContext).getDatastoreContext();
- doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(actorContext).getClientDispatcher();
- doReturn(getSystem()).when(actorContext).getActorSystem();
- doReturn(Dispatchers.DEFAULT_DISPATCHER_PATH).when(actorContext).getNotificationDispatcherPath();
- doReturn(getSystem().actorSelection(kit.getRef().path())).when(actorContext).actorSelection(
+ doReturn(DatastoreContext.newBuilder().build()).when(actorUtils).getDatastoreContext();
+ doReturn(getSystem().dispatchers().defaultGlobalDispatcher()).when(actorUtils).getClientDispatcher();
+ doReturn(getSystem()).when(actorUtils).getActorSystem();
+ doReturn(Dispatchers.DEFAULT_DISPATCHER_PATH).when(actorUtils).getNotificationDispatcherPath();
+ doReturn(getSystem().actorSelection(kit.getRef().path())).when(actorUtils).actorSelection(
- doReturn(kit.duration("5 seconds")).when(actorContext).getOperationDuration();
- doReturn(Futures.successful(kit.getRef())).when(actorContext).findLocalShardAsync(eq(shardName));
+ doReturn(kit.duration("5 seconds")).when(actorUtils).getOperationDuration();
+ doReturn(Futures.successful(kit.getRef())).when(actorUtils).findLocalShardAsync(eq(shardName));
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(
- actorContext, mockListener, YangInstanceIdentifier.of(TestModel.TEST_QNAME));
+ actorUtils, mockListener, YangInstanceIdentifier.of(TestModel.TEST_QNAME));
Answer<Future<Object>> answer = invocation -> {
proxy.close();
return Futures.successful((Object) new RegisterDataTreeNotificationListenerReply(kit.getRef()));
};
Answer<Future<Object>> answer = invocation -> {
proxy.close();
return Futures.successful((Object) new RegisterDataTreeNotificationListenerReply(kit.getRef()));
};
- doAnswer(answer).when(actorContext).executeOperationAsync(any(ActorRef.class), any(Object.class),
+ doAnswer(answer).when(actorUtils).executeOperationAsync(any(ActorRef.class), any(Object.class),
any(Timeout.class));
proxy.init(shardName);
any(Timeout.class));
proxy.init(shardName);
try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(testParameter, testName, false, shardName)) {
try (AbstractDataStore dataStore = testKit.setupAbstractDataStore(testParameter, testName, false, shardName)) {
- final Object result = dataStore.getActorContext().executeOperation(
- dataStore.getActorContext().getShardManager(), new FindLocalShard(shardName, true));
+ final Object result = dataStore.getActorUtils().executeOperation(
+ dataStore.getActorUtils().getShardManager(), new FindLocalShard(shardName, true));
assertTrue("Expected LocalShardFound. Actual: " + result, result instanceof LocalShardFound);
// Create the write Tx.
assertTrue("Expected LocalShardFound. Actual: " + result, result instanceof LocalShardFound);
// Create the write Tx.
followerDistributedDataStore = followerTestKit.setupAbstractDataStore(
testParameter, type, moduleShardsConfig, false, shards);
followerDistributedDataStore = followerTestKit.setupAbstractDataStore(
testParameter, type, moduleShardsConfig, false, shards);
- leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), shards);
+ leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(), shards);
leaderTestKit.waitForMembersUp("member-2");
followerTestKit.waitForMembersUp("member-1");
leaderTestKit.waitForMembersUp("member-2");
followerTestKit.waitForMembersUp("member-1");
TestKit.shutdownActorSystem(leaderSystem, true);
Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
TestKit.shutdownActorSystem(leaderSystem, true);
Cluster.get(followerSystem).leave(MEMBER_1_ADDRESS);
- followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorContext(), CARS);
+ followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorUtils(), CARS);
leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS);
leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS);
newMember1TestKit.setupAbstractDataStore(
testParameter, testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS)) {
newMember1TestKit.setupAbstractDataStore(
testParameter, testName, MODULE_SHARDS_CARS_ONLY_1_2, false, CARS)) {
- followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), CARS);
+ followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorUtils(), CARS);
// Write a car entry to the new leader - should switch to local Tx
// Write a car entry to the new leader - should switch to local Tx
@Test
public void testReadyLocalTransactionForwardedToLeader() throws Exception {
initDatastoresWithCars("testReadyLocalTransactionForwardedToLeader");
@Test
public void testReadyLocalTransactionForwardedToLeader() throws Exception {
initDatastoresWithCars("testReadyLocalTransactionForwardedToLeader");
- followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), "cars");
+ followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorUtils(), "cars");
final com.google.common.base.Optional<ActorRef> carsFollowerShard =
final com.google.common.base.Optional<ActorRef> carsFollowerShard =
- followerDistributedDataStore.getActorContext().findLocalShard("cars");
+ followerDistributedDataStore.getActorUtils().findLocalShard("cars");
assertTrue("Cars follower shard found", carsFollowerShard.isPresent());
final DataTree dataTree = new InMemoryDataTreeFactory().create(
assertTrue("Cars follower shard found", carsFollowerShard.isPresent());
final DataTree dataTree = new InMemoryDataTreeFactory().create(
assertEquals("Response type", ReadyTransactionReply.class, resp.getClass());
assertEquals("Response type", ReadyTransactionReply.class, resp.getClass());
- final ActorSelection txActor = leaderDistributedDataStore.getActorContext().actorSelection(
+ final ActorSelection txActor = leaderDistributedDataStore.getActorUtils().actorSelection(
((ReadyTransactionReply)resp).getCohortPath());
final Supplier<Short> versionSupplier = Mockito.mock(Supplier.class);
Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get();
ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(
((ReadyTransactionReply)resp).getCohortPath());
final Supplier<Short> versionSupplier = Mockito.mock(Supplier.class);
Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get();
ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(
- leaderDistributedDataStore.getActorContext(), Arrays.asList(
+ leaderDistributedDataStore.getActorUtils(), Arrays.asList(
new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), tx2);
cohort.canCommit().get(5, TimeUnit.SECONDS);
cohort.preCommit().get(5, TimeUnit.SECONDS);
new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), tx2);
cohort.canCommit().get(5, TimeUnit.SECONDS);
cohort.preCommit().get(5, TimeUnit.SECONDS);
@Test
public void testForwardedReadyTransactionForwardedToLeader() throws Exception {
initDatastoresWithCars("testForwardedReadyTransactionForwardedToLeader");
@Test
public void testForwardedReadyTransactionForwardedToLeader() throws Exception {
initDatastoresWithCars("testForwardedReadyTransactionForwardedToLeader");
- followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), "cars");
+ followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorUtils(), "cars");
final com.google.common.base.Optional<ActorRef> carsFollowerShard =
final com.google.common.base.Optional<ActorRef> carsFollowerShard =
- followerDistributedDataStore.getActorContext().findLocalShard("cars");
+ followerDistributedDataStore.getActorUtils().findLocalShard("cars");
assertTrue("Cars follower shard found", carsFollowerShard.isPresent());
carsFollowerShard.get().tell(GetShardDataTree.INSTANCE, followerTestKit.getRef());
assertTrue("Cars follower shard found", carsFollowerShard.isPresent());
carsFollowerShard.get().tell(GetShardDataTree.INSTANCE, followerTestKit.getRef());
assertEquals("Response type", ReadyTransactionReply.class, resp.getClass());
assertEquals("Response type", ReadyTransactionReply.class, resp.getClass());
- ActorSelection txActor = leaderDistributedDataStore.getActorContext().actorSelection(
+ ActorSelection txActor = leaderDistributedDataStore.getActorUtils().actorSelection(
((ReadyTransactionReply)resp).getCohortPath());
final Supplier<Short> versionSupplier = Mockito.mock(Supplier.class);
Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get();
final ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(
((ReadyTransactionReply)resp).getCohortPath());
final Supplier<Short> versionSupplier = Mockito.mock(Supplier.class);
Mockito.doReturn(DataStoreVersions.CURRENT_VERSION).when(versionSupplier).get();
final ThreePhaseCommitCohortProxy cohort = new ThreePhaseCommitCohortProxy(
- leaderDistributedDataStore.getActorContext(), Arrays.asList(
+ leaderDistributedDataStore.getActorUtils(), Arrays.asList(
new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), tx2);
cohort.canCommit().get(5, TimeUnit.SECONDS);
cohort.preCommit().get(5, TimeUnit.SECONDS);
new ThreePhaseCommitCohortProxy.CohortInfo(Futures.successful(txActor), versionSupplier)), tx2);
cohort.canCommit().get(5, TimeUnit.SECONDS);
cohort.preCommit().get(5, TimeUnit.SECONDS);
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName())
.shardElectionTimeoutFactor(10));
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName())
.shardElectionTimeoutFactor(10));
- leaderTestKit.waitUntilNoLeader(leaderDistributedDataStore.getActorContext(), "cars");
+ leaderTestKit.waitUntilNoLeader(leaderDistributedDataStore.getActorUtils(), "cars");
// Submit all tx's - the messages should get queued for retry.
// Submit all tx's - the messages should get queued for retry.
sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
.customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1));
sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder
.customRaftPolicyImplementation(null).shardElectionTimeoutFactor(1));
- IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorContext(), "cars")
+ IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorUtils(), "cars")
.tell(TimeoutNow.INSTANCE, ActorRef.noSender());
.tell(TimeoutNow.INSTANCE, ActorRef.noSender());
- IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorContext(), "people")
+ IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorUtils(), "people")
.tell(TimeoutNow.INSTANCE, ActorRef.noSender());
followerTestKit.doCommit(writeTx1CanCommit, writeTx1Cohort);
.tell(TimeoutNow.INSTANCE, ActorRef.noSender());
followerTestKit.doCommit(writeTx1CanCommit, writeTx1Cohort);
.shardElectionTimeoutFactor(100));
final FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
.shardElectionTimeoutFactor(100));
final FiniteDuration duration = FiniteDuration.create(5, TimeUnit.SECONDS);
- final Future<ActorRef> future = leaderDistributedDataStore.getActorContext().findLocalShardAsync("cars");
+ final Future<ActorRef> future = leaderDistributedDataStore.getActorUtils().findLocalShardAsync("cars");
final ActorRef leaderActor = Await.result(future, duration);
final Future<Boolean> stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE);
final ActorRef leaderActor = Await.result(future, duration);
final Future<Boolean> stopFuture = Patterns.gracefulStop(leaderActor, duration, Shutdown.INSTANCE);
successWriteTx.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
// Stop the follower
successWriteTx.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
// Stop the follower
- followerTestKit.watch(followerDistributedDataStore.getActorContext().getShardManager());
+ followerTestKit.watch(followerDistributedDataStore.getActorUtils().getShardManager());
followerDistributedDataStore.close();
followerDistributedDataStore.close();
- followerTestKit.expectTerminated(followerDistributedDataStore.getActorContext().getShardManager());
+ followerTestKit.expectTerminated(followerDistributedDataStore.getActorUtils().getShardManager());
// Submit the preIsolatedLeaderWriteTx so it's pending
final DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready();
// Submit the preIsolatedLeaderWriteTx so it's pending
final DOMStoreThreePhaseCommitCohort preIsolatedLeaderTxCohort = preIsolatedLeaderWriteTx.ready();
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.access.concepts.FrontendType;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
import org.opendaylight.controller.cluster.access.concepts.FrontendType;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import scala.concurrent.duration.FiniteDuration;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import scala.concurrent.duration.FiniteDuration;
private static SchemaContext SCHEMA_CONTEXT;
@Mock
private static SchemaContext SCHEMA_CONTEXT;
@Mock
- private ActorContext actorContext;
+ private ActorUtils actorUtils;
@Mock
private DatastoreContext datastoreContext;
@Mock
private DatastoreContext datastoreContext;
public void setUp() {
MockitoAnnotations.initMocks(this);
public void setUp() {
MockitoAnnotations.initMocks(this);
- doReturn(SCHEMA_CONTEXT).when(actorContext).getSchemaContext();
- doReturn(DatastoreContext.newBuilder().build()).when(actorContext).getDatastoreContext();
+ doReturn(SCHEMA_CONTEXT).when(actorUtils).getSchemaContext();
+ doReturn(DatastoreContext.newBuilder().build()).when(actorUtils).getDatastoreContext();
}
@Test
public void testRateLimitingUsedInReadWriteTxCreation() {
}
@Test
public void testRateLimitingUsedInReadWriteTxCreation() {
- try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorContext, UNKNOWN_ID)) {
+ try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorUtils, UNKNOWN_ID)) {
distributedDataStore.newReadWriteTransaction();
distributedDataStore.newReadWriteTransaction();
- verify(actorContext, times(1)).acquireTxCreationPermit();
+ verify(actorUtils, times(1)).acquireTxCreationPermit();
}
}
@Test
public void testRateLimitingUsedInWriteOnlyTxCreation() {
}
}
@Test
public void testRateLimitingUsedInWriteOnlyTxCreation() {
- try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorContext, UNKNOWN_ID)) {
+ try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorUtils, UNKNOWN_ID)) {
distributedDataStore.newWriteOnlyTransaction();
distributedDataStore.newWriteOnlyTransaction();
- verify(actorContext, times(1)).acquireTxCreationPermit();
+ verify(actorUtils, times(1)).acquireTxCreationPermit();
}
}
@Test
public void testRateLimitingNotUsedInReadOnlyTxCreation() {
}
}
@Test
public void testRateLimitingNotUsedInReadOnlyTxCreation() {
- try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorContext, UNKNOWN_ID)) {
+ try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorUtils, UNKNOWN_ID)) {
distributedDataStore.newReadOnlyTransaction();
distributedDataStore.newReadOnlyTransaction();
distributedDataStore.newReadOnlyTransaction();
distributedDataStore.newReadOnlyTransaction();
distributedDataStore.newReadOnlyTransaction();
distributedDataStore.newReadOnlyTransaction();
- verify(actorContext, times(0)).acquireTxCreationPermit();
+ verify(actorUtils, times(0)).acquireTxCreationPermit();
}
}
@Test
public void testWaitTillReadyBlocking() {
}
}
@Test
public void testWaitTillReadyBlocking() {
- doReturn(datastoreContext).when(actorContext).getDatastoreContext();
+ doReturn(datastoreContext).when(actorUtils).getDatastoreContext();
doReturn(shardElectionTimeout).when(datastoreContext).getShardLeaderElectionTimeout();
doReturn(FiniteDuration.apply(50, TimeUnit.MILLISECONDS)).when(shardElectionTimeout).duration();
doReturn(shardElectionTimeout).when(datastoreContext).getShardLeaderElectionTimeout();
doReturn(FiniteDuration.apply(50, TimeUnit.MILLISECONDS)).when(shardElectionTimeout).duration();
- try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorContext, UNKNOWN_ID)) {
+ try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorUtils, UNKNOWN_ID)) {
long start = System.currentTimeMillis();
long start = System.currentTimeMillis();
@Test
public void testWaitTillReadyCountDown() {
@Test
public void testWaitTillReadyCountDown() {
- try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorContext, UNKNOWN_ID)) {
- doReturn(datastoreContext).when(actorContext).getDatastoreContext();
+ try (DistributedDataStore distributedDataStore = new DistributedDataStore(actorUtils, UNKNOWN_ID)) {
+ doReturn(datastoreContext).when(actorUtils).getDatastoreContext();
doReturn(shardElectionTimeout).when(datastoreContext).getShardLeaderElectionTimeout();
doReturn(FiniteDuration.apply(5000, TimeUnit.MILLISECONDS)).when(shardElectionTimeout).duration();
doReturn(shardElectionTimeout).when(datastoreContext).getShardLeaderElectionTimeout();
doReturn(FiniteDuration.apply(5000, TimeUnit.MILLISECONDS)).when(shardElectionTimeout).duration();
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
dataStore.onGlobalContextUpdated(schemaContext);
if (waitUntilLeader) {
dataStore.onGlobalContextUpdated(schemaContext);
if (waitUntilLeader) {
- waitUntilLeader(dataStore.getActorContext(), shardNames);
+ waitUntilLeader(dataStore.getActorUtils(), shardNames);
}
datastoreContextBuilder = DatastoreContext.newBuilderFrom(datastoreContext);
}
datastoreContextBuilder = DatastoreContext.newBuilderFrom(datastoreContext);
- public void waitUntilLeader(final ActorContext actorContext, final String... shardNames) {
+ public void waitUntilLeader(final ActorUtils actorUtils, final String... shardNames) {
for (String shardName: shardNames) {
for (String shardName: shardNames) {
- ActorRef shard = findLocalShard(actorContext, shardName);
+ ActorRef shard = findLocalShard(actorUtils, shardName);
assertNotNull("Shard was not created for " + shardName, shard);
assertNotNull("Shard was not created for " + shardName, shard);
- public void waitUntilNoLeader(final ActorContext actorContext, final String... shardNames) {
+ public void waitUntilNoLeader(final ActorUtils actorUtils, final String... shardNames) {
for (String shardName: shardNames) {
for (String shardName: shardNames) {
- ActorRef shard = findLocalShard(actorContext, shardName);
+ ActorRef shard = findLocalShard(actorUtils, shardName);
assertNotNull("No local shard found for " + shardName, shard);
waitUntilNoLeader(shard);
assertNotNull("No local shard found for " + shardName, shard);
waitUntilNoLeader(shard);
fail("Member(s) " + otherMembersSet + " are not Up");
}
fail("Member(s) " + otherMembersSet + " are not Up");
}
- public static ActorRef findLocalShard(final ActorContext actorContext, final String shardName) {
+ public static ActorRef findLocalShard(final ActorUtils actorUtils, final String shardName) {
ActorRef shard = null;
for (int i = 0; i < 20 * 5 && shard == null; i++) {
Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
ActorRef shard = null;
for (int i = 0; i < 20 * 5 && shard == null; i++) {
Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
- com.google.common.base.Optional<ActorRef> shardReply = actorContext.findLocalShard(shardName);
+ com.google.common.base.Optional<ActorRef> shardReply = actorUtils.findLocalShard(shardName);
if (shardReply.isPresent()) {
shard = shardReply.get();
}
if (shardReply.isPresent()) {
shard = shardReply.get();
}
- public static void waitUntilShardIsDown(final ActorContext actorContext, final String shardName) {
+ public static void waitUntilShardIsDown(final ActorUtils actorUtils, final String shardName) {
for (int i = 0; i < 20 * 5 ; i++) {
LOG.debug("Waiting for shard down {}", shardName);
Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
for (int i = 0; i < 20 * 5 ; i++) {
LOG.debug("Waiting for shard down {}", shardName);
Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
- com.google.common.base.Optional<ActorRef> shardReply = actorContext.findLocalShard(shardName);
+ com.google.common.base.Optional<ActorRef> shardReply = actorUtils.findLocalShard(shardName);
if (!shardReply.isPresent()) {
return;
}
if (!shardReply.isPresent()) {
return;
}
public static void verifyShardStats(final AbstractDataStore datastore, final String shardName,
final ShardStatsVerifier verifier) throws Exception {
public static void verifyShardStats(final AbstractDataStore datastore, final String shardName,
final ShardStatsVerifier verifier) throws Exception {
- ActorContext actorContext = datastore.getActorContext();
+ ActorUtils actorUtils = datastore.getActorUtils();
- Future<ActorRef> future = actorContext.findLocalShardAsync(shardName);
+ Future<ActorRef> future = actorUtils.findLocalShardAsync(shardName);
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
AssertionError lastError = null;
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
AssertionError lastError = null;
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
- ShardStats shardStats = (ShardStats)actorContext
+ ShardStats shardStats = (ShardStats)actorUtils
.executeOperation(shardActor, Shard.GET_SHARD_MBEAN_MESSAGE);
try {
.executeOperation(shardActor, Shard.GET_SHARD_MBEAN_MESSAGE);
try {
public static void verifyShardState(final AbstractDataStore datastore, final String shardName,
final Consumer<OnDemandShardState> verifier) throws Exception {
public static void verifyShardState(final AbstractDataStore datastore, final String shardName,
final Consumer<OnDemandShardState> verifier) throws Exception {
- ActorContext actorContext = datastore.getActorContext();
+ ActorUtils actorUtils = datastore.getActorUtils();
- Future<ActorRef> future = actorContext.findLocalShardAsync(shardName);
+ Future<ActorRef> future = actorUtils.findLocalShardAsync(shardName);
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
AssertionError lastError = null;
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
AssertionError lastError = null;
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
- OnDemandShardState shardState = (OnDemandShardState)actorContext
+ OnDemandShardState shardState = (OnDemandShardState)actorUtils
.executeOperation(shardActor, GetOnDemandRaftState.INSTANCE);
try {
.executeOperation(shardActor, GetOnDemandRaftState.INSTANCE);
try {
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
public static void verifyRaftState(final AbstractDataStore datastore, final String shardName,
final RaftStateVerifier verifier) throws Exception {
public static void verifyRaftState(final AbstractDataStore datastore, final String shardName,
final RaftStateVerifier verifier) throws Exception {
- ActorContext actorContext = datastore.getActorContext();
+ ActorUtils actorUtils = datastore.getActorUtils();
- Future<ActorRef> future = actorContext.findLocalShardAsync(shardName);
+ Future<ActorRef> future = actorUtils.findLocalShardAsync(shardName);
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
AssertionError lastError = null;
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
AssertionError lastError = null;
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
- OnDemandRaftState raftState = (OnDemandRaftState)actorContext
+ OnDemandRaftState raftState = (OnDemandRaftState)actorUtils
.executeOperation(shardActor, GetOnDemandRaftState.INSTANCE);
try {
.executeOperation(shardActor, GetOnDemandRaftState.INSTANCE);
try {
final Set<String> peerIds = Sets.newHashSet();
for (String p: peerMemberNames) {
peerIds.add(ShardIdentifier.create(shardName, MemberName.forName(p),
final Set<String> peerIds = Sets.newHashSet();
for (String p: peerMemberNames) {
peerIds.add(ShardIdentifier.create(shardName, MemberName.forName(p),
- datastore.getActorContext().getDataStoreName()).toString());
+ datastore.getActorUtils().getDataStoreName()).toString());
}
verifyRaftState(datastore, shardName, raftState -> assertEquals("Peers for shard " + shardName, peerIds,
}
verifyRaftState(datastore, shardName, raftState -> assertEquals("Peers for shard " + shardName, peerIds,
public static void verifyNoShardPresent(final AbstractDataStore datastore, final String shardName) {
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
public static void verifyNoShardPresent(final AbstractDataStore datastore, final String shardName) {
Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
- Optional<ActorRef> shardReply = datastore.getActorContext().findLocalShard(shardName);
+ Optional<ActorRef> shardReply = datastore.getActorUtils().findLocalShard(shardName);
if (!shardReply.isPresent()) {
return;
}
if (!shardReply.isPresent()) {
return;
}
import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
private OperationLimiter limiter;
private RemoteTransactionContext txContext;
private OperationLimiter limiter;
private RemoteTransactionContext txContext;
- private ActorContext actorContext;
+ private ActorUtils actorUtils;
private TestKit kit;
@Before
public void before() {
kit = new TestKit(getSystem());
private TestKit kit;
@Before
public void before() {
kit = new TestKit(getSystem());
- actorContext = Mockito.spy(new ActorContext(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
+ actorUtils = Mockito.spy(new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class),
mock(Configuration.class)));
limiter = new OperationLimiter(TX_ID, 4, 0);
mock(Configuration.class)));
limiter = new OperationLimiter(TX_ID, 4, 0);
- txContext = new RemoteTransactionContext(TX_ID, actorContext.actorSelection(kit.getRef().path()), actorContext,
+ txContext = new RemoteTransactionContext(TX_ID, actorUtils.actorSelection(kit.getRef().path()), actorUtils,
DataStoreVersions.CURRENT_VERSION, limiter);
txContext.operationHandOffComplete();
}
DataStoreVersions.CURRENT_VERSION, limiter);
txContext.operationHandOffComplete();
}
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
static class TestException extends RuntimeException {
}
static class TestException extends RuntimeException {
}
- private ActorContext actorContext;
+ private ActorUtils actorUtils;
@Mock
private Timer commitTimer;
@Mock
private Timer commitTimer;
public void setUp() {
MockitoAnnotations.initMocks(this);
public void setUp() {
MockitoAnnotations.initMocks(this);
- actorContext = new ActorContext(getSystem(), actorFactory.createActor(Props.create(DoNothingActor.class)),
+ actorUtils = new ActorUtils(getSystem(), actorFactory.createActor(Props.create(DoNothingActor.class)),
new MockClusterWrapper(), new MockConfiguration(), DatastoreContext.newBuilder().build(),
new PrimaryShardInfoFutureCache()) {
@Override
new MockClusterWrapper(), new MockConfiguration(), DatastoreContext.newBuilder().build(),
new PrimaryShardInfoFutureCache()) {
@Override
@Test
public void testCanCommitYesWithOneCohort() throws Exception {
@Test
public void testCanCommitYesWithOneCohort() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.yes(CURRENT_VERSION)))), tx);
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.yes(CURRENT_VERSION)))), tx);
@Test
public void testCanCommitNoWithOneCohort() throws Exception {
@Test
public void testCanCommitNoWithOneCohort() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.no(CURRENT_VERSION)))), tx);
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.no(CURRENT_VERSION)))), tx);
CanCommitTransactionReply.yes(CURRENT_VERSION))),
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.yes(CURRENT_VERSION))));
CanCommitTransactionReply.yes(CURRENT_VERSION))),
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.yes(CURRENT_VERSION))));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), true);
verifyCohortActors();
verifyCanCommit(proxy.canCommit(), true);
verifyCohortActors();
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.no(CURRENT_VERSION))),
newCohortInfo(new CohortActor.Builder(tx)));
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(
CanCommitTransactionReply.no(CURRENT_VERSION))),
newCohortInfo(new CohortActor.Builder(tx)));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), false);
verifyCohortActors();
verifyCanCommit(proxy.canCommit(), false);
verifyCohortActors();
@Test(expected = TestException.class)
public void testCanCommitWithExceptionFailure() throws Exception {
@Test(expected = TestException.class)
public void testCanCommitWithExceptionFailure() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(new TestException()))), tx);
propagateExecutionExceptionCause(proxy.canCommit());
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit(new TestException()))), tx);
propagateExecutionExceptionCause(proxy.canCommit());
@Test(expected = IllegalArgumentException.class)
public void testCanCommitWithInvalidResponseType() throws Exception {
@Test(expected = IllegalArgumentException.class)
public void testCanCommitWithInvalidResponseType() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit("invalid"))), tx);
propagateExecutionExceptionCause(proxy.canCommit());
newCohortInfo(new CohortActor.Builder(tx).expectCanCommit("invalid"))), tx);
propagateExecutionExceptionCause(proxy.canCommit());
newCohortInfo(new CohortActor.Builder(tx)),
newCohortInfoWithFailedFuture(new TestException()),
newCohortInfo(new CohortActor.Builder(tx)));
newCohortInfo(new CohortActor.Builder(tx)),
newCohortInfoWithFailedFuture(new TestException()),
newCohortInfo(new CohortActor.Builder(tx)));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
propagateExecutionExceptionCause(proxy.canCommit());
}
propagateExecutionExceptionCause(proxy.canCommit());
}
newCohortInfo(
new CohortActor.Builder(tx).expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION))
.expectCommit(CommitTransactionReply.instance(CURRENT_VERSION))));
newCohortInfo(
new CohortActor.Builder(tx).expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION))
.expectCommit(CommitTransactionReply.instance(CURRENT_VERSION))));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), true);
verifySuccessfulFuture(proxy.preCommit());
verifyCanCommit(proxy.canCommit(), true);
verifySuccessfulFuture(proxy.preCommit());
newCohortInfo(
new CohortActor.Builder(tx).expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION))
.expectCommit(new TestException())));
newCohortInfo(
new CohortActor.Builder(tx).expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION))
.expectCommit(new TestException())));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifyCanCommit(proxy.canCommit(), true);
verifySuccessfulFuture(proxy.preCommit());
verifyCanCommit(proxy.canCommit(), true);
verifySuccessfulFuture(proxy.preCommit());
@Test(expected = IllegalArgumentException.class)
public void testCommitWithInvalidResponseType() throws Exception {
@Test(expected = IllegalArgumentException.class)
public void testCommitWithInvalidResponseType() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext,
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils,
Arrays.asList(newCohortInfo(new CohortActor.Builder(tx)
.expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).expectCommit("invalid"))), tx);
Arrays.asList(newCohortInfo(new CohortActor.Builder(tx)
.expectCanCommit(CanCommitTransactionReply.yes(CURRENT_VERSION)).expectCommit("invalid"))), tx);
@Test
public void testAbort() throws Exception {
@Test
public void testAbort() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectAbort(
AbortTransactionReply.instance(CURRENT_VERSION)))), tx);
newCohortInfo(new CohortActor.Builder(tx).expectAbort(
AbortTransactionReply.instance(CURRENT_VERSION)))), tx);
@Test
public void testAbortWithFailure() throws Exception {
@Test
public void testAbortWithFailure() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, Arrays.asList(
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, Arrays.asList(
newCohortInfo(new CohortActor.Builder(tx).expectAbort(new RuntimeException("mock")))), tx);
// The exception should not get propagated.
newCohortInfo(new CohortActor.Builder(tx).expectAbort(new RuntimeException("mock")))), tx);
// The exception should not get propagated.
public void testAbortWithFailedCohortFuture() throws Exception {
List<CohortInfo> cohorts = Arrays.asList(
newCohortInfoWithFailedFuture(new TestException()), newCohortInfo(new CohortActor.Builder(tx)));
public void testAbortWithFailedCohortFuture() throws Exception {
List<CohortInfo> cohorts = Arrays.asList(
newCohortInfoWithFailedFuture(new TestException()), newCohortInfo(new CohortActor.Builder(tx)));
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext, cohorts, tx);
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils, cohorts, tx);
verifySuccessfulFuture(proxy.abort());
verifyCohortActors();
verifySuccessfulFuture(proxy.abort());
verifyCohortActors();
@Test
public void testWithNoCohorts() throws Exception {
@Test
public void testWithNoCohorts() throws Exception {
- ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorContext,
+ ThreePhaseCommitCohortProxy proxy = new ThreePhaseCommitCohortProxy(actorUtils,
Collections.<CohortInfo>emptyList(), tx);
verifyCanCommit(proxy.canCommit(), true);
Collections.<CohortInfo>emptyList(), tx);
verifyCanCommit(proxy.canCommit(), true);
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
public class TransactionContextWrapperTest {
@Mock
public class TransactionContextWrapperTest {
@Mock
- private ActorContext actorContext;
+ private ActorUtils actorUtils;
@Mock
private TransactionContext transactionContext;
@Mock
private TransactionContext transactionContext;
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
- doReturn(DatastoreContext.newBuilder().build()).when(actorContext).getDatastoreContext();
+ doReturn(DatastoreContext.newBuilder().build()).when(actorUtils).getDatastoreContext();
transactionContextWrapper = new TransactionContextWrapper(MockIdentifiers.transactionIdentifier(
transactionContextWrapper = new TransactionContextWrapper(MockIdentifiers.transactionIdentifier(
- TransactionContextWrapperTest.class, "mock"), actorContext, "mock");
+ TransactionContextWrapperTest.class, "mock"), actorUtils, "mock");
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
/**
* Unit tests for TransactionRateLimitingCallback.
/**
* Unit tests for TransactionRateLimitingCallback.
public class TransactionRateLimitingCallbackTest {
@Mock
public class TransactionRateLimitingCallbackTest {
@Mock
- ActorContext mockContext;
+ ActorUtils mockContext;
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
- doReturn(mockTimer).when(mockContext).getOperationTimer(ActorContext.COMMIT);
+ doReturn(mockTimer).when(mockContext).getOperationTimer(ActorUtils.COMMIT);
callback = new TransactionRateLimitingCallback(mockContext);
TransactionRateLimitingCallback.setTicker(mockTicker);
}
callback = new TransactionRateLimitingCallback(mockContext);
TransactionRateLimitingCallback.setTicker(mockTicker);
}
}
private static DistributedEntityOwnershipService newOwnershipService(final AbstractDataStore datastore) {
}
private static DistributedEntityOwnershipService newOwnershipService(final AbstractDataStore datastore) {
- return DistributedEntityOwnershipService.start(datastore.getActorContext(),
+ return DistributedEntityOwnershipService.start(datastore.getActorUtils(),
EntityOwnerSelectionStrategyConfig.newBuilder().build());
}
EntityOwnerSelectionStrategyConfig.newBuilder().build());
}
final DOMEntityOwnershipService follower2EntityOwnershipService =
newOwnershipService(follower2Node.configDataStore());
final DOMEntityOwnershipService follower2EntityOwnershipService =
newOwnershipService(follower2Node.configDataStore());
- leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
+ leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorUtils(), ENTITY_OWNERSHIP_SHARD_NAME);
leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
leaderEntityOwnershipService.registerListener(ENTITY_TYPE2, leaderMockListener2);
leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
leaderEntityOwnershipService.registerListener(ENTITY_TYPE2, leaderMockListener2);
final DOMEntityOwnershipService follower2EntityOwnershipService =
newOwnershipService(follower2Node.configDataStore());
final DOMEntityOwnershipService follower2EntityOwnershipService =
newOwnershipService(follower2Node.configDataStore());
- leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
+ leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorUtils(), ENTITY_OWNERSHIP_SHARD_NAME);
// Register follower1 candidate for entity1 and verify it becomes owner
// Register follower1 candidate for entity1 and verify it becomes owner
// Re-enable elections on all remaining followers so one becomes the new leader
// Re-enable elections on all remaining followers so one becomes the new leader
- ActorRef follower1Shard = IntegrationTestKit.findLocalShard(follower1Node.configDataStore().getActorContext(),
+ ActorRef follower1Shard = IntegrationTestKit.findLocalShard(follower1Node.configDataStore().getActorUtils(),
ENTITY_OWNERSHIP_SHARD_NAME);
follower1Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
ENTITY_OWNERSHIP_SHARD_NAME);
follower1Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
- ActorRef follower2Shard = IntegrationTestKit.findLocalShard(follower2Node.configDataStore().getActorContext(),
+ ActorRef follower2Shard = IntegrationTestKit.findLocalShard(follower2Node.configDataStore().getActorUtils(),
ENTITY_OWNERSHIP_SHARD_NAME);
follower2Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
ENTITY_OWNERSHIP_SHARD_NAME);
follower2Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
newOwnershipService(follower3Node.configDataStore());
newOwnershipService(follower4Node.configDataStore());
newOwnershipService(follower3Node.configDataStore());
newOwnershipService(follower4Node.configDataStore());
- leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
+ leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorUtils(), ENTITY_OWNERSHIP_SHARD_NAME);
// Register follower1 candidate for entity1 and verify it becomes owner
// Register follower1 candidate for entity1 and verify it becomes owner
// Re-enable elections on all remaining followers so one becomes the new leader
// Re-enable elections on all remaining followers so one becomes the new leader
- ActorRef follower1Shard = IntegrationTestKit.findLocalShard(follower1Node.configDataStore().getActorContext(),
+ ActorRef follower1Shard = IntegrationTestKit.findLocalShard(follower1Node.configDataStore().getActorUtils(),
ENTITY_OWNERSHIP_SHARD_NAME);
follower1Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
ENTITY_OWNERSHIP_SHARD_NAME);
follower1Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
- ActorRef follower2Shard = IntegrationTestKit.findLocalShard(follower2Node.configDataStore().getActorContext(),
+ ActorRef follower2Shard = IntegrationTestKit.findLocalShard(follower2Node.configDataStore().getActorUtils(),
ENTITY_OWNERSHIP_SHARD_NAME);
follower2Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
ENTITY_OWNERSHIP_SHARD_NAME);
follower2Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
- ActorRef follower4Shard = IntegrationTestKit.findLocalShard(follower4Node.configDataStore().getActorContext(),
+ ActorRef follower4Shard = IntegrationTestKit.findLocalShard(follower4Node.configDataStore().getActorUtils(),
ENTITY_OWNERSHIP_SHARD_NAME);
follower4Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
ENTITY_OWNERSHIP_SHARD_NAME);
follower4Shard.tell(DatastoreContext.newBuilderFrom(followerDatastoreContextBuilder.build())
.customRaftPolicyImplementation(null).build(), ActorRef.noSender());
final DOMEntityOwnershipService follower2EntityOwnershipService =
newOwnershipService(follower2Node.configDataStore());
final DOMEntityOwnershipService follower2EntityOwnershipService =
newOwnershipService(follower2Node.configDataStore());
- leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
+ leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorUtils(), ENTITY_OWNERSHIP_SHARD_NAME);
leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
follower1EntityOwnershipService.registerListener(ENTITY_TYPE1, follower1MockListener);
leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
follower1EntityOwnershipService.registerListener(ENTITY_TYPE1, follower1MockListener);
AbstractDataStore leaderDistributedDataStore = leaderNode.configDataStore();
final DOMEntityOwnershipService leaderEntityOwnershipService = newOwnershipService(leaderDistributedDataStore);
AbstractDataStore leaderDistributedDataStore = leaderNode.configDataStore();
final DOMEntityOwnershipService leaderEntityOwnershipService = newOwnershipService(leaderDistributedDataStore);
- leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
+ leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorUtils(), ENTITY_OWNERSHIP_SHARD_NAME);
MemberNode follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name)
.moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false)
MemberNode follower1Node = MemberNode.builder(memberNodes).akkaConfig("Member2").testName(name)
.moduleShardsConfig(moduleShardsConfig).schemaContext(SCHEMA_CONTEXT).createOperDatastore(false)
// Add replica in follower1
AddShardReplica addReplica = new AddShardReplica(ENTITY_OWNERSHIP_SHARD_NAME);
// Add replica in follower1
AddShardReplica addReplica = new AddShardReplica(ENTITY_OWNERSHIP_SHARD_NAME);
- follower1DistributedDataStore.getActorContext().getShardManager().tell(addReplica,
+ follower1DistributedDataStore.getActorUtils().getShardManager().tell(addReplica,
follower1Node.kit().getRef());
Object reply = follower1Node.kit().expectMsgAnyClassOf(follower1Node.kit().duration("5 sec"),
Success.class, Failure.class);
follower1Node.kit().getRef());
Object reply = follower1Node.kit().expectMsgAnyClassOf(follower1Node.kit().duration("5 sec"),
Success.class, Failure.class);
newOwnershipService(follower1Node.configDataStore());
newOwnershipService(follower2Node.configDataStore());
newOwnershipService(follower1Node.configDataStore());
newOwnershipService(follower2Node.configDataStore());
- leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
+ leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorUtils(), ENTITY_OWNERSHIP_SHARD_NAME);
// Register leader candidate for entity1 and verify it becomes owner
// Register leader candidate for entity1 and verify it becomes owner
newOwnershipService(follower1Node.configDataStore());
newOwnershipService(follower2Node.configDataStore());
newOwnershipService(follower1Node.configDataStore());
newOwnershipService(follower2Node.configDataStore());
- leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
+ leaderNode.kit().waitUntilLeader(leaderNode.configDataStore().getActorUtils(), ENTITY_OWNERSHIP_SHARD_NAME);
// Register leader candidate for entity1 and verify it becomes owner
// Register leader candidate for entity1 and verify it becomes owner
newOwnershipService(member5FollowerNode.configDataStore());
newOwnershipService(member1LeaderNode.configDataStore());
newOwnershipService(member5FollowerNode.configDataStore());
newOwnershipService(member1LeaderNode.configDataStore());
- member1LeaderNode.kit().waitUntilLeader(member1LeaderNode.configDataStore().getActorContext(),
+ member1LeaderNode.kit().waitUntilLeader(member1LeaderNode.configDataStore().getActorUtils(),
ENTITY_OWNERSHIP_SHARD_NAME);
// Make member4 and member5 non-voting
ENTITY_OWNERSHIP_SHARD_NAME);
// Make member4 and member5 non-voting
- Future<Object> future = Patterns.ask(leaderDistributedDataStore.getActorContext().getShardManager(),
+ Future<Object> future = Patterns.ask(leaderDistributedDataStore.getActorUtils().getShardManager(),
new ChangeShardMembersVotingStatus(ENTITY_OWNERSHIP_SHARD_NAME,
ImmutableMap.of("member-4", Boolean.FALSE, "member-5", Boolean.FALSE)),
new Timeout(10, TimeUnit.SECONDS));
new ChangeShardMembersVotingStatus(ENTITY_OWNERSHIP_SHARD_NAME,
ImmutableMap.of("member-4", Boolean.FALSE, "member-5", Boolean.FALSE)),
new Timeout(10, TimeUnit.SECONDS));
// Switch member4 and member5 back to voting and member3 non-voting. This should result in member4 and member5
// to become entity owners.
// Switch member4 and member5 back to voting and member3 non-voting. This should result in member4 and member5
// to become entity owners.
- future = Patterns.ask(leaderDistributedDataStore.getActorContext().getShardManager(),
+ future = Patterns.ask(leaderDistributedDataStore.getActorUtils().getShardManager(),
new ChangeShardMembersVotingStatus(ENTITY_OWNERSHIP_SHARD_NAME,
ImmutableMap.of("member-3", Boolean.FALSE, "member-4", Boolean.TRUE, "member-5", Boolean.TRUE)),
new Timeout(10, TimeUnit.SECONDS));
new ChangeShardMembersVotingStatus(ENTITY_OWNERSHIP_SHARD_NAME,
ImmutableMap.of("member-3", Boolean.FALSE, "member-4", Boolean.TRUE, "member-5", Boolean.TRUE)),
new Timeout(10, TimeUnit.SECONDS));
@Test
public void testEntityOwnershipShardCreated() throws Exception {
@Test
public void testEntityOwnershipShardCreated() throws Exception {
- DistributedEntityOwnershipService service = DistributedEntityOwnershipService.start(dataStore.getActorContext(),
+ DistributedEntityOwnershipService service = DistributedEntityOwnershipService.start(dataStore.getActorUtils(),
EntityOwnerSelectionStrategyConfig.newBuilder().build());
EntityOwnerSelectionStrategyConfig.newBuilder().build());
- Future<ActorRef> future = dataStore.getActorContext().findLocalShardAsync(
+ Future<ActorRef> future = dataStore.getActorUtils().findLocalShardAsync(
DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME);
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
assertNotNull(DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME + " not found", shardActor);
DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME);
ActorRef shardActor = Await.result(future, FiniteDuration.create(10, TimeUnit.SECONDS));
assertNotNull(DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME + " not found", shardActor);
@Test
public void testRegisterCandidate() throws Exception {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
@Test
public void testRegisterCandidate() throws Exception {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
- dataStore.getActorContext(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
+ dataStore.getActorUtils(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
YangInstanceIdentifier entityId = YangInstanceIdentifier.of(QNAME);
DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
YangInstanceIdentifier entityId = YangInstanceIdentifier.of(QNAME);
DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
verifyRegisterCandidateLocal(service, entity);
verifyEntityOwnershipCandidateRegistration(entity, reg);
verifyEntityCandidate(service.getLocalEntityOwnershipShard(), ENTITY_TYPE, entityId,
verifyRegisterCandidateLocal(service, entity);
verifyEntityOwnershipCandidateRegistration(entity, reg);
verifyEntityCandidate(service.getLocalEntityOwnershipShard(), ENTITY_TYPE, entityId,
- dataStore.getActorContext().getCurrentMemberName().getName());
+ dataStore.getActorUtils().getCurrentMemberName().getName());
// Register the same entity - should throw exception
// Register the same entity - should throw exception
verifyEntityOwnershipCandidateRegistration(entity2, reg2);
verifyEntityCandidate(service.getLocalEntityOwnershipShard(), ENTITY_TYPE2, entityId,
verifyEntityOwnershipCandidateRegistration(entity2, reg2);
verifyEntityCandidate(service.getLocalEntityOwnershipShard(), ENTITY_TYPE2, entityId,
- dataStore.getActorContext().getCurrentMemberName().getName());
+ dataStore.getActorUtils().getCurrentMemberName().getName());
@Test
public void testCloseCandidateRegistration() throws Exception {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
@Test
public void testCloseCandidateRegistration() throws Exception {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
- dataStore.getActorContext(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
+ dataStore.getActorUtils(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
DOMEntity entity = new DOMEntity(ENTITY_TYPE, YangInstanceIdentifier.of(QNAME));
DOMEntity entity = new DOMEntity(ENTITY_TYPE, YangInstanceIdentifier.of(QNAME));
@Test
public void testListenerRegistration() {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
@Test
public void testListenerRegistration() {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
- dataStore.getActorContext(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
+ dataStore.getActorUtils(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
YangInstanceIdentifier entityId = YangInstanceIdentifier.of(QNAME);
DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
YangInstanceIdentifier entityId = YangInstanceIdentifier.of(QNAME);
DOMEntity entity = new DOMEntity(ENTITY_TYPE, entityId);
@Test
public void testGetOwnershipState() throws Exception {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
@Test
public void testGetOwnershipState() throws Exception {
DistributedEntityOwnershipService service = spy(DistributedEntityOwnershipService.start(
- dataStore.getActorContext(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
+ dataStore.getActorUtils(), EntityOwnerSelectionStrategyConfig.newBuilder().build()));
final Shard mockShard = Mockito.mock(Shard.class);
ShardDataTree shardDataTree = new ShardDataTree(mockShard, SchemaContextHelper.entityOwners(),
final Shard mockShard = Mockito.mock(Shard.class);
ShardDataTree shardDataTree = new ShardDataTree(mockShard, SchemaContextHelper.entityOwners(),
@Test
public void testIsCandidateRegistered() throws CandidateAlreadyRegisteredException {
@Test
public void testIsCandidateRegistered() throws CandidateAlreadyRegisteredException {
- DistributedEntityOwnershipService service = DistributedEntityOwnershipService.start(dataStore.getActorContext(),
+ DistributedEntityOwnershipService service = DistributedEntityOwnershipService.start(dataStore.getActorUtils(),
EntityOwnerSelectionStrategyConfig.newBuilder().build());
final DOMEntity test = new DOMEntity("test-type", "test");
EntityOwnerSelectionStrategyConfig.newBuilder().build());
final DOMEntity test = new DOMEntity("test-type", "test");
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
-public class ActorContextTest extends AbstractActorTest {
+public class ActorUtilsTest extends AbstractActorTest {
- static final Logger LOG = LoggerFactory.getLogger(ActorContextTest.class);
+ static final Logger LOG = LoggerFactory.getLogger(ActorUtilsTest.class);
private static class TestMessage {
}
private static class TestMessage {
}
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
- ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef,
mock(ClusterWrapper.class), mock(Configuration.class));
mock(ClusterWrapper.class), mock(Configuration.class));
- Optional<ActorRef> out = actorContext.findLocalShard("default");
+ Optional<ActorRef> out = actorUtils.findLocalShard("default");
assertEquals(shardActorRef, out.get());
assertEquals(shardActorRef, out.get());
public void testFindLocalShardWithShardNotFound() {
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(false, null));
public void testFindLocalShardWithShardNotFound() {
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(false, null));
- ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef, mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef, mock(ClusterWrapper.class),
mock(Configuration.class));
mock(Configuration.class));
- Optional<ActorRef> out = actorContext.findLocalShard("default");
+ Optional<ActorRef> out = actorUtils.findLocalShard("default");
assertFalse(out.isPresent());
}
assertFalse(out.isPresent());
}
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
- ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef,
mock(ClusterWrapper.class), mock(Configuration.class));
mock(ClusterWrapper.class), mock(Configuration.class));
- ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
+ ActorSelection actor = actorUtils.actorSelection(shardActorRef.path());
- Object out = actorContext.executeOperation(actor, "hello");
+ Object out = actorUtils.executeOperation(actor, "hello");
assertEquals("hello", out);
}
assertEquals("hello", out);
}
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
- ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef,
mock(ClusterWrapper.class), mock(Configuration.class));
mock(ClusterWrapper.class), mock(Configuration.class));
- ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
+ ActorSelection actor = actorUtils.actorSelection(shardActorRef.path());
- Future<Object> future = actorContext.executeOperationAsync(actor, "hello");
+ Future<Object> future = actorUtils.executeOperationAsync(actor, "hello");
Object result = Await.result(future, FiniteDuration.create(3, TimeUnit.SECONDS));
assertEquals("Result", "hello", result);
Object result = Await.result(future, FiniteDuration.create(3, TimeUnit.SECONDS));
assertEquals("Result", "hello", result);
@Test
public void testIsPathLocal() {
MockClusterWrapper clusterWrapper = new MockClusterWrapper();
@Test
public void testIsPathLocal() {
MockClusterWrapper clusterWrapper = new MockClusterWrapper();
- ActorContext actorContext = null;
+ ActorUtils actorUtils = null;
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertFalse(actorContext.isPathLocal(null));
- assertFalse(actorContext.isPathLocal(""));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertFalse(actorUtils.isPathLocal(null));
+ assertFalse(actorUtils.isPathLocal(""));
clusterWrapper.setSelfAddress(null);
clusterWrapper.setSelfAddress(null);
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertFalse(actorContext.isPathLocal(""));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertFalse(actorUtils.isPathLocal(""));
// even if the path is in local format, match the primary path (first 3 elements) and return true
clusterWrapper.setSelfAddress(new Address("akka", "test"));
// even if the path is in local format, match the primary path (first 3 elements) and return true
clusterWrapper.setSelfAddress(new Address("akka", "test"));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertTrue(actorContext.isPathLocal("akka://test/user/$a"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertTrue(actorUtils.isPathLocal("akka://test/user/$a"));
clusterWrapper.setSelfAddress(new Address("akka", "test"));
clusterWrapper.setSelfAddress(new Address("akka", "test"));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertTrue(actorContext.isPathLocal("akka://test/user/$a"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertTrue(actorUtils.isPathLocal("akka://test/user/$a"));
clusterWrapper.setSelfAddress(new Address("akka", "test"));
clusterWrapper.setSelfAddress(new Address("akka", "test"));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertTrue(actorContext.isPathLocal("akka://test/user/token2/token3/$a"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertTrue(actorUtils.isPathLocal("akka://test/user/token2/token3/$a"));
// self address of remote format,but Tx path local format.
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
// self address of remote format,but Tx path local format.
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertTrue(actorContext.isPathLocal("akka://system/user/shardmanager/shard/transaction"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertTrue(actorUtils.isPathLocal("akka://system/user/shardmanager/shard/transaction"));
// self address of local format,but Tx path remote format.
clusterWrapper.setSelfAddress(new Address("akka", "system"));
// self address of local format,but Tx path remote format.
clusterWrapper.setSelfAddress(new Address("akka", "system"));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertFalse(actorContext.isPathLocal("akka://system@127.0.0.1:2550/user/shardmanager/shard/transaction"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertFalse(actorUtils.isPathLocal("akka://system@127.0.0.1:2550/user/shardmanager/shard/transaction"));
//local path but not same
clusterWrapper.setSelfAddress(new Address("akka", "test"));
//local path but not same
clusterWrapper.setSelfAddress(new Address("akka", "test"));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertTrue(actorContext.isPathLocal("akka://test1/user/$a"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertTrue(actorUtils.isPathLocal("akka://test1/user/$a"));
//ip and port same
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
//ip and port same
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertTrue(actorContext.isPathLocal("akka://system@127.0.0.1:2550/"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertTrue(actorUtils.isPathLocal("akka://system@127.0.0.1:2550/"));
// forward-slash missing in address
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
// forward-slash missing in address
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertFalse(actorContext.isPathLocal("akka://system@127.0.0.1:2550"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertFalse(actorUtils.isPathLocal("akka://system@127.0.0.1:2550"));
//ips differ
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
//ips differ
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertFalse(actorContext.isPathLocal("akka://system@127.1.0.1:2550/"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertFalse(actorUtils.isPathLocal("akka://system@127.1.0.1:2550/"));
//ports differ
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
//ports differ
clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
- actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
- assertFalse(actorContext.isPathLocal("akka://system@127.0.0.1:2551/"));
+ actorUtils = new ActorUtils(getSystem(), null, clusterWrapper, mock(Configuration.class));
+ assertFalse(actorUtils.isPathLocal("akka://system@127.0.0.1:2551/"));
}
@Test
public void testClientDispatcherIsGlobalDispatcher() {
}
@Test
public void testClientDispatcherIsGlobalDispatcher() {
- ActorContext actorContext = new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
mock(Configuration.class), DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
mock(Configuration.class), DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
- assertEquals(getSystem().dispatchers().defaultGlobalDispatcher(), actorContext.getClientDispatcher());
+ assertEquals(getSystem().dispatchers().defaultGlobalDispatcher(), actorUtils.getClientDispatcher());
ActorSystem actorSystem = ActorSystem.create("with-custom-dispatchers",
ConfigFactory.load("application-with-custom-dispatchers.conf"));
ActorSystem actorSystem = ActorSystem.create("with-custom-dispatchers",
ConfigFactory.load("application-with-custom-dispatchers.conf"));
- ActorContext actorContext = new ActorContext(actorSystem, mock(ActorRef.class), mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(actorSystem, mock(ActorRef.class), mock(ClusterWrapper.class),
mock(Configuration.class), DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
mock(Configuration.class), DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
- assertNotEquals(actorSystem.dispatchers().defaultGlobalDispatcher(), actorContext.getClientDispatcher());
+ assertNotEquals(actorSystem.dispatchers().defaultGlobalDispatcher(), actorUtils.getClientDispatcher());
actorSystem.terminate();
}
actorSystem.terminate();
}
@Test
public void testSetDatastoreContext() {
final TestKit testKit = new TestKit(getSystem());
@Test
public void testSetDatastoreContext() {
final TestKit testKit = new TestKit(getSystem());
- ActorContext actorContext = new ActorContext(getSystem(), testKit.getRef(),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), testKit.getRef(),
mock(ClusterWrapper.class), mock(Configuration.class), DatastoreContext.newBuilder()
.operationTimeoutInSeconds(5).shardTransactionCommitTimeoutInSeconds(7).build(),
new PrimaryShardInfoFutureCache());
mock(ClusterWrapper.class), mock(Configuration.class), DatastoreContext.newBuilder()
.operationTimeoutInSeconds(5).shardTransactionCommitTimeoutInSeconds(7).build(),
new PrimaryShardInfoFutureCache());
- assertEquals("getOperationDuration", 5, actorContext.getOperationDuration().toSeconds());
+ assertEquals("getOperationDuration", 5, actorUtils.getOperationDuration().toSeconds());
assertEquals("getTransactionCommitOperationTimeout", 7,
assertEquals("getTransactionCommitOperationTimeout", 7,
- actorContext.getTransactionCommitOperationTimeout().duration().toSeconds());
+ actorUtils.getTransactionCommitOperationTimeout().duration().toSeconds());
DatastoreContext newContext = DatastoreContext.newBuilder().operationTimeoutInSeconds(6)
.shardTransactionCommitTimeoutInSeconds(8).build();
DatastoreContext newContext = DatastoreContext.newBuilder().operationTimeoutInSeconds(6)
.shardTransactionCommitTimeoutInSeconds(8).build();
DatastoreContextFactory mockContextFactory = mock(DatastoreContextFactory.class);
Mockito.doReturn(newContext).when(mockContextFactory).getBaseDatastoreContext();
DatastoreContextFactory mockContextFactory = mock(DatastoreContextFactory.class);
Mockito.doReturn(newContext).when(mockContextFactory).getBaseDatastoreContext();
- actorContext.setDatastoreContext(mockContextFactory);
+ actorUtils.setDatastoreContext(mockContextFactory);
testKit.expectMsgClass(Duration.ofSeconds(5), DatastoreContextFactory.class);
testKit.expectMsgClass(Duration.ofSeconds(5), DatastoreContextFactory.class);
- Assert.assertSame("getDatastoreContext", newContext, actorContext.getDatastoreContext());
+ Assert.assertSame("getDatastoreContext", newContext, actorUtils.getDatastoreContext());
- assertEquals("getOperationDuration", 6, actorContext.getOperationDuration().toSeconds());
+ assertEquals("getOperationDuration", 6, actorUtils.getOperationDuration().toSeconds());
assertEquals("getTransactionCommitOperationTimeout", 8,
assertEquals("getTransactionCommitOperationTimeout", 8,
- actorContext.getTransactionCommitOperationTimeout().duration().toSeconds());
+ actorUtils.getTransactionCommitOperationTimeout().duration().toSeconds());
final String expPrimaryPath = "akka://test-system/find-primary-shard";
final short expPrimaryVersion = DataStoreVersions.CURRENT_VERSION;
final String expPrimaryPath = "akka://test-system/find-primary-shard";
final short expPrimaryVersion = DataStoreVersions.CURRENT_VERSION;
- ActorContext actorContext = new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManager, mock(ClusterWrapper.class),
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
@Override
protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
@Override
protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
- Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
+ Future<PrimaryShardInfo> foobar = actorUtils.findPrimaryShardAsync("foobar");
PrimaryShardInfo actual = Await.result(foobar, FiniteDuration.apply(5000, TimeUnit.MILLISECONDS));
assertNotNull(actual);
PrimaryShardInfo actual = Await.result(foobar, FiniteDuration.apply(5000, TimeUnit.MILLISECONDS));
assertNotNull(actual);
expPrimaryPath.endsWith(actual.getPrimaryShardActor().pathString()));
assertEquals("getPrimaryShardVersion", expPrimaryVersion, actual.getPrimaryShardVersion());
expPrimaryPath.endsWith(actual.getPrimaryShardActor().pathString()));
assertEquals("getPrimaryShardVersion", expPrimaryVersion, actual.getPrimaryShardVersion());
- Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
+ Future<PrimaryShardInfo> cached = actorUtils.getPrimaryShardInfoCache().getIfPresent("foobar");
PrimaryShardInfo cachedInfo = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
assertEquals(cachedInfo, actual);
PrimaryShardInfo cachedInfo = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
assertEquals(cachedInfo, actual);
- actorContext.getPrimaryShardInfoCache().remove("foobar");
+ actorUtils.getPrimaryShardInfoCache().remove("foobar");
- cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
+ cached = actorUtils.getPrimaryShardInfoCache().getIfPresent("foobar");
final DataTree mockDataTree = Mockito.mock(DataTree.class);
final String expPrimaryPath = "akka://test-system/find-primary-shard";
final DataTree mockDataTree = Mockito.mock(DataTree.class);
final String expPrimaryPath = "akka://test-system/find-primary-shard";
- ActorContext actorContext = new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManager, mock(ClusterWrapper.class),
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
@Override
protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
@Override
protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
- Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
+ Future<PrimaryShardInfo> foobar = actorUtils.findPrimaryShardAsync("foobar");
PrimaryShardInfo actual = Await.result(foobar, FiniteDuration.apply(5000, TimeUnit.MILLISECONDS));
assertNotNull(actual);
PrimaryShardInfo actual = Await.result(foobar, FiniteDuration.apply(5000, TimeUnit.MILLISECONDS));
assertNotNull(actual);
expPrimaryPath.endsWith(actual.getPrimaryShardActor().pathString()));
assertEquals("getPrimaryShardVersion", DataStoreVersions.CURRENT_VERSION, actual.getPrimaryShardVersion());
expPrimaryPath.endsWith(actual.getPrimaryShardActor().pathString()));
assertEquals("getPrimaryShardVersion", DataStoreVersions.CURRENT_VERSION, actual.getPrimaryShardVersion());
- Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
+ Future<PrimaryShardInfo> cached = actorUtils.getPrimaryShardInfoCache().getIfPresent("foobar");
PrimaryShardInfo cachedInfo = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
assertEquals(cachedInfo, actual);
PrimaryShardInfo cachedInfo = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
assertEquals(cachedInfo, actual);
- actorContext.getPrimaryShardInfoCache().remove("foobar");
+ actorUtils.getPrimaryShardInfoCache().remove("foobar");
- cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
+ cached = actorUtils.getPrimaryShardInfoCache().getIfPresent("foobar");
.logicalStoreType(LogicalDatastoreType.CONFIGURATION)
.shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
.logicalStoreType(LogicalDatastoreType.CONFIGURATION)
.shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
- ActorContext actorContext =
- new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManager, mock(ClusterWrapper.class),
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
- @Override
- protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
- return Futures.successful(expectedException);
- }
- };
+ @Override
+ protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
+ return Futures.successful(expectedException);
+ }
+ };
- Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
+ Future<PrimaryShardInfo> foobar = actorUtils.findPrimaryShardAsync("foobar");
try {
Await.result(foobar, FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
try {
Await.result(foobar, FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
- Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
+ Future<PrimaryShardInfo> cached = actorUtils.getPrimaryShardInfoCache().getIfPresent("foobar");
doReturn(Sets.newLinkedHashSet(Arrays.asList("shard1", "shard2", "shard3"))).when(mockConfig)
.getAllShardNames();
doReturn(Sets.newLinkedHashSet(Arrays.asList("shard1", "shard2", "shard3"))).when(mockConfig)
.getAllShardNames();
- ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
+ ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef,
mock(ClusterWrapper.class), mockConfig,
DatastoreContext.newBuilder().shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(),
new PrimaryShardInfoFutureCache());
mock(ClusterWrapper.class), mockConfig,
DatastoreContext.newBuilder().shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(),
new PrimaryShardInfoFutureCache());
- actorContext.broadcast(v -> new TestMessage(), TestMessage.class);
+ actorUtils.broadcast(v -> new TestMessage(), TestMessage.class);
MessageCollectorActor.expectFirstMatching(shardActorRef1, TestMessage.class);
MessageCollectorActor.expectFirstMatching(shardActorRef2, TestMessage.class);
MessageCollectorActor.expectFirstMatching(shardActorRef1, TestMessage.class);
MessageCollectorActor.expectFirstMatching(shardActorRef2, TestMessage.class);
public class TransactionRateLimiterTest {
@Mock
public class TransactionRateLimiterTest {
@Mock
- public ActorContext actorContext;
+ public ActorUtils actorUtils;
@Mock
public DatastoreContext datastoreContext;
@Mock
public DatastoreContext datastoreContext;
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
@Before
public void setUp() {
MockitoAnnotations.initMocks(this);
- doReturn(datastoreContext).when(actorContext).getDatastoreContext();
+ doReturn(datastoreContext).when(actorUtils).getDatastoreContext();
doReturn(30).when(datastoreContext).getShardTransactionCommitTimeoutInSeconds();
doReturn(100L).when(datastoreContext).getTransactionCreationInitialRateLimit();
doReturn(30).when(datastoreContext).getShardTransactionCommitTimeoutInSeconds();
doReturn(100L).when(datastoreContext).getTransactionCreationInitialRateLimit();
- doReturn(commitTimer).when(actorContext).getOperationTimer("commit");
+ doReturn(commitTimer).when(actorUtils).getOperationTimer("commit");
doReturn(commitTimerContext).when(commitTimer).time();
doReturn(commitSnapshot).when(commitTimer).getSnapshot();
}
doReturn(commitTimerContext).when(commitTimer).time();
doReturn(commitSnapshot).when(commitTimer).getSnapshot();
}
doReturn(TimeUnit.MILLISECONDS.toNanos(i) * 1D).when(commitSnapshot).getValue(i * 0.1);
}
doReturn(TimeUnit.MILLISECONDS.toNanos(i) * 1D).when(commitSnapshot).getValue(i * 0.1);
}
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
doReturn(TimeUnit.MILLISECONDS.toNanos(0) * 1D).when(commitSnapshot).getValue(0.1);
doReturn(TimeUnit.MILLISECONDS.toNanos(0) * 1D).when(commitSnapshot).getValue(0.1);
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
// ten seconds
doReturn(TimeUnit.MILLISECONDS.toNanos(10000) * 1D).when(commitSnapshot).getValue(1.0);
// ten seconds
doReturn(TimeUnit.MILLISECONDS.toNanos(10000) * 1D).when(commitSnapshot).getValue(1.0);
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
doReturn(TimeUnit.MILLISECONDS.toNanos(10000) * 1D).when(commitSnapshot).getValue(i * 0.1);
}
doReturn(TimeUnit.MILLISECONDS.toNanos(10000) * 1D).when(commitSnapshot).getValue(i * 0.1);
}
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
doReturn(TimeUnit.MILLISECONDS.toNanos(100) * 1D).when(commitSnapshot).getValue(0.9);
doReturn(TimeUnit.MILLISECONDS.toNanos(200) * 1D).when(commitSnapshot).getValue(1.0);
doReturn(TimeUnit.MILLISECONDS.toNanos(100) * 1D).when(commitSnapshot).getValue(0.9);
doReturn(TimeUnit.MILLISECONDS.toNanos(200) * 1D).when(commitSnapshot).getValue(1.0);
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
Timer.Context operationalCommitTimerContext = mock(Timer.Context.class);
Snapshot operationalCommitSnapshot = mock(Snapshot.class);
Timer.Context operationalCommitTimerContext = mock(Timer.Context.class);
Snapshot operationalCommitSnapshot = mock(Snapshot.class);
- doReturn(operationalCommitTimer).when(actorContext).getOperationTimer("operational", "commit");
+ doReturn(operationalCommitTimer).when(actorUtils).getOperationTimer("operational", "commit");
doReturn(operationalCommitTimerContext).when(operationalCommitTimer).time();
doReturn(operationalCommitSnapshot).when(operationalCommitTimer).getSnapshot();
doReturn(operationalCommitTimerContext).when(operationalCommitTimer).time();
doReturn(operationalCommitSnapshot).when(operationalCommitTimer).getSnapshot();
DatastoreContext.getGlobalDatastoreNames().add("config");
DatastoreContext.getGlobalDatastoreNames().add("operational");
DatastoreContext.getGlobalDatastoreNames().add("config");
DatastoreContext.getGlobalDatastoreNames().add("operational");
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
doReturn(TimeUnit.SECONDS.toNanos(1) * 1D).when(commitSnapshot).getValue(i * 0.1);
}
doReturn(TimeUnit.SECONDS.toNanos(1) * 1D).when(commitSnapshot).getValue(i * 0.1);
}
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
StopWatch watch = new StopWatch();
StopWatch watch = new StopWatch();
doReturn(TimeUnit.MILLISECONDS.toNanos(100) * 1D).when(commitSnapshot).getValue(0.9);
doReturn(TimeUnit.MILLISECONDS.toNanos(200) * 1D).when(commitSnapshot).getValue(1.0);
doReturn(TimeUnit.MILLISECONDS.toNanos(100) * 1D).when(commitSnapshot).getValue(0.9);
doReturn(TimeUnit.MILLISECONDS.toNanos(200) * 1D).when(commitSnapshot).getValue(1.0);
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
doReturn(TimeUnit.MILLISECONDS.toNanos(100) * 1D).when(commitSnapshot).getValue(0.9);
doReturn(TimeUnit.MILLISECONDS.toNanos(200) * 1D).when(commitSnapshot).getValue(1.0);
doReturn(TimeUnit.MILLISECONDS.toNanos(100) * 1D).when(commitSnapshot).getValue(0.9);
doReturn(TimeUnit.MILLISECONDS.toNanos(200) * 1D).when(commitSnapshot).getValue(1.0);
- TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorContext);
+ TransactionRateLimiter rateLimiter = new TransactionRateLimiter(actorUtils);
rateLimiter.setAcquireCount(Long.MAX_VALUE - 1);
rateLimiter.setPollOnCount(Long.MAX_VALUE);
rateLimiter.setAcquireCount(Long.MAX_VALUE - 1);
rateLimiter.setPollOnCount(Long.MAX_VALUE);
import org.opendaylight.controller.cluster.datastore.AbstractActorTest;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
import org.opendaylight.controller.cluster.datastore.AbstractActorTest;
import org.opendaylight.controller.cluster.datastore.DatastoreContext;
import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.dom.api.LeaderLocation;
import org.opendaylight.controller.cluster.dom.api.LeaderLocationListener;
import org.opendaylight.controller.cluster.dom.api.LeaderLocationListenerRegistration;
import org.opendaylight.controller.cluster.dom.api.LeaderLocation;
import org.opendaylight.controller.cluster.dom.api.LeaderLocationListener;
import org.opendaylight.controller.cluster.dom.api.LeaderLocationListenerRegistration;
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
private CDSShardAccessImpl shardAccess;
new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
private CDSShardAccessImpl shardAccess;
- private ActorContext context;
+ private ActorUtils context;
@Before
public void setUp() {
@Before
public void setUp() {
- context = mock(ActorContext.class);
+ context = mock(ActorUtils.class);
final DatastoreContext datastoreContext = DatastoreContext.newBuilder().build();
doReturn(Optional.of(getSystem().deadLetters())).when(context).findLocalShard(any());
doReturn(datastoreContext).when(context).getDatastoreContext();
final DatastoreContext datastoreContext = DatastoreContext.newBuilder().build();
doReturn(Optional.of(getSystem().deadLetters())).when(context).findLocalShard(any());
doReturn(datastoreContext).when(context).getDatastoreContext();
import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
public void testClientTransaction() throws Exception {
final DistributedDataStore distributedDataStore = mock(DistributedDataStore.class);
public void testClientTransaction() throws Exception {
final DistributedDataStore distributedDataStore = mock(DistributedDataStore.class);
- final ActorContext context = mock(ActorContext.class);
- doReturn(context).when(distributedDataStore).getActorContext();
+ final ActorUtils context = mock(ActorUtils.class);
+ doReturn(context).when(distributedDataStore).getActorUtils();
doReturn(SchemaContextHelper.full()).when(context).getSchemaContext();
final DistributedShardFrontend rootShard = new DistributedShardFrontend(distributedDataStore, client, ROOT);
doReturn(SchemaContextHelper.full()).when(context).getSchemaContext();
final DistributedShardFrontend rootShard = new DistributedShardFrontend(distributedDataStore, client, ROOT);
LOG.info("Initializing leader DistributedShardedDOMDataTree");
leaderShardFactory.init();
LOG.info("Initializing leader DistributedShardedDOMDataTree");
leaderShardFactory.init();
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
- leaderTestKit.waitUntilLeader(leaderOperDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderOperDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
LOG.info("Initializing follower DistributedShardedDOMDataTree");
ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
LOG.info("Initializing follower DistributedShardedDOMDataTree");
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
- final ActorRef leaderShardManager = leaderConfigDatastore.getActorContext().getShardManager();
+ final ActorRef leaderShardManager = leaderConfigDatastore.getActorUtils().getShardManager();
- assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
- assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
final Set<String> peers = new HashSet<>();
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
final Set<String> peers = new HashSet<>();
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
- findLocalShard(followerConfigDatastore.getActorContext(),
+ findLocalShard(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
final Set<String> peers = new HashSet<>();
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
final Set<String> peers = new HashSet<>();
Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
// check leader has local shards
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
// check leader has local shards
- assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
- assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
- assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
- assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
// check follower has local shards
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
// check follower has local shards
- assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
- assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
- assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
- assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
LOG.debug("Closing registrations");
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
LOG.debug("Closing registrations");
reg3.close().toCompletableFuture().get();
reg4.close().toCompletableFuture().get();
reg3.close().toCompletableFuture().get();
reg4.close().toCompletableFuture().get();
- waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
- waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
- waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
- waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
LOG.debug("All leader shards gone");
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
LOG.debug("All leader shards gone");
- waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
- waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
- waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
- waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
LOG.debug("All follower shards gone");
ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
LOG.debug("All follower shards gone");
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
- assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
- assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
+ assertNotNull(findLocalShard(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(leaderConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
- waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
+ waitUntilShardIsDown(followerConfigDatastore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
}
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
}
import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.dom.api.CDSDataTreeProducer;
import org.opendaylight.controller.cluster.dom.api.CDSShardAccess;
leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
final String shardName = ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier());
LOG.debug("Creating distributed datastore client for shard {}", shardName);
final String shardName = ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier());
LOG.debug("Creating distributed datastore client for shard {}", shardName);
- final ActorContext actorContext = leaderDistributedDataStore.getActorContext();
+ final ActorUtils actorUtils = leaderDistributedDataStore.getActorUtils();
final Props distributedDataStoreClientProps =
final Props distributedDataStoreClientProps =
- SimpleDataStoreClientActor.props(actorContext.getCurrentMemberName(),
- "Shard-" + shardName, actorContext, shardName);
+ SimpleDataStoreClientActor.props(actorUtils.getCurrentMemberName(), "Shard-" + shardName, actorUtils,
+ shardName);
final ActorRef clientActor = leaderSystem.actorOf(distributedDataStoreClientProps);
final DataStoreClient distributedDataStoreClient = SimpleDataStoreClientActor
final ActorRef clientActor = leaderSystem.actorOf(distributedDataStoreClientProps);
final DataStoreClient distributedDataStoreClient = SimpleDataStoreClientActor
leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
LOG.warn("Got after waiting for nonleader");
ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
LOG.warn("Got after waiting for nonleader");
- final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
+ final ActorRef leaderShardManager = leaderDistributedDataStore.getActorUtils().getShardManager();
- leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
final YangInstanceIdentifier oid1 = getOuterListIdFor(0);
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
final YangInstanceIdentifier oid1 = getOuterListIdFor(0);
leaderShardFactory.createDistributedShard(outerListPath, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
leaderShardFactory.createDistributedShard(outerListPath, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(outerListPath.getRootIdentifier()));
final DOMDataTreeProducer shardProducer = leaderShardFactory.createProducer(
ClusterUtils.getCleanShardName(outerListPath.getRootIdentifier()));
final DOMDataTreeProducer shardProducer = leaderShardFactory.createProducer(
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
- assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(),
+ assertNotNull(findLocalShard(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- waitUntilShardIsDown(leaderDistributedDataStore.getActorContext(),
+ waitUntilShardIsDown(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
}
}
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
}
}
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)),
DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
- leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
+ leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
- assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(),
+ assertNotNull(findLocalShard(leaderDistributedDataStore.getActorUtils(),
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
import org.opendaylight.controller.cluster.databroker.actors.dds.SimpleDataStoreClientActor;
import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
import org.opendaylight.controller.cluster.sharding.DistributedShardFactory;
import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
import org.opendaylight.controller.cluster.sharding.DistributedShardFactory;
private <T> SettableFuture<RpcResult<T>> shutdownShardGracefully(final String shardName, final T success) {
final SettableFuture<RpcResult<T>> rpcResult = SettableFuture.create();
private <T> SettableFuture<RpcResult<T>> shutdownShardGracefully(final String shardName, final T success) {
final SettableFuture<RpcResult<T>> rpcResult = SettableFuture.create();
- final ActorContext context = configDataStore.getActorContext();
+ final ActorUtils context = configDataStore.getActorUtils();
long timeoutInMS = Math.max(context.getDatastoreContext().getShardRaftConfig()
.getElectionTimeOutInterval().$times(3).toMillis(), 10000);
long timeoutInMS = Math.max(context.getDatastoreContext().getShardRaftConfig()
.getElectionTimeOutInterval().$times(3).toMillis(), 10000);
final String shardName = ClusterUtils.getCleanShardName(ProduceTransactionsHandler.ID_INTS_YID);
LOG.debug("Creating distributed datastore client for shard {}", shardName);
final String shardName = ClusterUtils.getCleanShardName(ProduceTransactionsHandler.ID_INTS_YID);
LOG.debug("Creating distributed datastore client for shard {}", shardName);
- final ActorContext actorContext = configDataStore.getActorContext();
+ final ActorUtils actorUtils = configDataStore.getActorUtils();
final Props distributedDataStoreClientProps =
final Props distributedDataStoreClientProps =
- SimpleDataStoreClientActor.props(actorContext.getCurrentMemberName(),
- "Shard-" + shardName, actorContext, shardName);
+ SimpleDataStoreClientActor.props(actorUtils.getCurrentMemberName(),
+ "Shard-" + shardName, actorUtils, shardName);
final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
final DataStoreClient distributedDataStoreClient;
final ActorRef clientActor = actorSystem.actorOf(distributedDataStoreClientProps);
final DataStoreClient distributedDataStoreClient;
Code Review / controller.git / commitdiff