import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot;
-import org.opendaylight.controller.cluster.datastore.persisted.ShardDataTreeSnapshot;
+import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState;
import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener;
import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListenerReply;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
-import org.opendaylight.controller.cluster.raft.Snapshot;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
-import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker;
import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
+import org.opendaylight.yangtools.concepts.Identifier;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
- "testRegisterChangeListener-DataChangeListener");
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener,
+ TestModel.TEST_PATH), "testRegisterChangeListener-DataChangeListener");
shard.tell(new RegisterChangeListener(TestModel.TEST_PATH, dclActor,
AsyncDataBroker.DataChangeScope.BASE, true), getRef());
setupInMemorySnapshotStore();
+ final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, path),
"testRegisterChangeListenerWhenNotLeaderInitially-DataChangeListener");
final TestActorRef<Shard> shard = actorFactory.createTestActor(
new ShardTestKit(getSystem()) {
{
- final YangInstanceIdentifier path = TestModel.TEST_PATH;
-
// Wait until the shard receives the first ElectionTimeout
// message.
assertEquals("Got first ElectionTimeout", true, onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
shard.tell(new UpdateSchemaContext(SchemaContextHelper.full()), ActorRef.noSender());
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
- "testRegisterDataTreeChangeListener-DataTreeChangeListener");
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener,
+ TestModel.TEST_PATH), "testRegisterDataTreeChangeListener-DataTreeChangeListener");
shard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, false), getRef());
setupInMemorySnapshotStore();
+ final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener, path),
"testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration-DataChangeListener");
final TestActorRef<Shard> shard = actorFactory.createTestActor(
Props.create(new DelegatingShardCreator(creator)).withDispatcher(Dispatchers.DefaultDispatcherId()),
"testDataTreeChangeListenerNotifiedWhenNotTheLeaderOnRegistration");
- final YangInstanceIdentifier path = TestModel.TEST_PATH;
-
new ShardTestKit(getSystem()) {
{
assertEquals("Got first ElectionTimeout", true, onFirstElectionTimeout.await(5, TimeUnit.SECONDS));
final YangInstanceIdentifier root = YangInstanceIdentifier.EMPTY;
final NormalizedNode<?,?> expected = readStore(store, root);
- final Snapshot snapshot = Snapshot.create(new MetadataShardDataTreeSnapshot(expected).serialize(),
- Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4);
+ final Snapshot snapshot = Snapshot.create(
+ new ShardSnapshotState(new MetadataShardDataTreeSnapshot(expected)),
+ Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4, -1, null, null);
shard.tell(new ApplySnapshot(snapshot), ActorRef.noSender());
final DataTree store = InMemoryDataTreeFactory.getInstance().create(TreeType.OPERATIONAL);
store.setSchemaContext(SCHEMA_CONTEXT);
- writeToStore(store, TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
-
- final NormalizedNode<?, ?> root = readStore(store, YangInstanceIdentifier.EMPTY);
- final Snapshot snapshot = Snapshot.create(new MetadataShardDataTreeSnapshot(root).serialize(),
- Collections.<ReplicatedLogEntry>emptyList(), 1, 2, 3, 4);
-
- shard.tell(new ApplySnapshot(snapshot), ActorRef.noSender());
final DataTreeModification writeMod = store.takeSnapshot().newModification();
final ContainerNode node = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
writeMod.ready();
final TransactionIdentifier tx = nextTransactionId();
- final ApplyState applyState = new ApplyState(null, tx,
- new SimpleReplicatedLogEntry(1, 2, payloadForModification(store, writeMod, tx)));
-
- shard.tell(applyState, shard);
+ shard.underlyingActor().applyState(null, null, payloadForModification(store, writeMod, tx));
final Stopwatch sw = Stopwatch.createStarted();
while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
{
final Creator<Shard> creator = () -> new Shard(newShardBuilder()) {
@Override
- void persistPayload(final TransactionIdentifier transactionId, final Payload payload,
- boolean batchHint) {
+ void persistPayload(final Identifier id, final Payload payload,
+ final boolean batchHint) {
// Simulate an AbortTransaction message occurring during
// replication, after
// persisting and before finishing the commit to the
// in-memory store.
- doAbortTransaction(transactionId, null);
- super.persistPayload(transactionId, payload, batchHint);
+ doAbortTransaction(id, null);
+ super.persistPayload(id, payload, batchHint);
}
};
private void verifySnapshot(final Snapshot snapshot, final NormalizedNode<?, ?> expectedRoot)
throws IOException {
- final NormalizedNode<?, ?> actual = ShardDataTreeSnapshot.deserialize(snapshot.getState()).getRootNode()
- .get();
+ final NormalizedNode<?, ?> actual = ((ShardSnapshotState)snapshot.getState()).getSnapshot()
+ .getRootNode().get();
assertEquals("Root node", expectedRoot, actual);
}
};
dataStoreContextBuilder.shardElectionTimeoutFactor(1000)
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
+ final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, path),
actorFactory.generateActorId(testName + "-DataChangeListener"));
setupInMemorySnapshotStore();
waitUntilNoLeader(shard);
- final YangInstanceIdentifier path = TestModel.TEST_PATH;
-
shard.tell(new RegisterChangeListener(path, dclActor, AsyncDataBroker.DataChangeScope.BASE, true),
getRef());
final RegisterChangeListenerReply reply = expectMsgClass(duration("5 seconds"),
final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataChangeListener listener = new MockDataChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataChangeListener.props(listener, path),
actorFactory.generateActorId(testName + "-DataChangeListener"));
followerShard.tell(
.customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
- actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener,
+ TestModel.TEST_PATH), actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
setupInMemorySnapshotStore();
final YangInstanceIdentifier path = TestModel.TEST_PATH;
final MockDataTreeChangeListener listener = new MockDataTreeChangeListener(1);
- final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener),
+ final ActorRef dclActor = actorFactory.createActor(DataTreeChangeListenerActor.props(listener, path),
actorFactory.generateActorId(testName + "-DataTreeChangeListener"));
followerShard.tell(new RegisterDataTreeChangeListener(TestModel.TEST_PATH, dclActor, true), getRef());