import akka.testkit.TestActorRef;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
-import org.junit.Assert;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
+import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.DataExists;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
-
+import scala.concurrent.duration.Duration;
import java.util.Collections;
-
+import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
ShardIdentifier.builder().memberName("member-1")
.shardName("inventory").type("config").build();
+ private DatastoreContext datastoreContext = DatastoreContext.newBuilder().build();
+
+ private final ShardStats shardStats = new ShardStats(SHARD_IDENTIFIER.toString(), "DataStore");
- static {
+ @BeforeClass
+ public static void staticSetup() {
store.onGlobalContextUpdated(testSchemaContext);
}
+ private ActorRef createShard(){
+ return getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
+ Collections.EMPTY_MAP, datastoreContext, TestModel.createTestContext()));
+ }
+
@Test
public void testOnReceiveReadData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props(store.newReadOnlyTransaction(), shard, testSchemaContext);
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject = getSystem().actorOf(props, "testReadData");
new Within(duration("1 seconds")) {
@Test
public void testOnReceiveReadDataWhenDataNotFound() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props( store.newReadOnlyTransaction(), shard, testSchemaContext);
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props( store.newReadOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject = getSystem().actorOf(props, "testReadDataWhenDataNotFound");
new Within(duration("1 seconds")) {
@Test
public void testOnReceiveDataExistsPositive() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props(store.newReadOnlyTransaction(), shard, testSchemaContext);
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject = getSystem().actorOf(props, "testDataExistsPositive");
new Within(duration("1 seconds")) {
@Test
public void testOnReceiveDataExistsNegative() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
- Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props(store.newReadOnlyTransaction(), shard, testSchemaContext);
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject = getSystem().actorOf(props, "testDataExistsNegative");
new Within(duration("1 seconds")) {
private void assertModification(final ActorRef subject,
final Class<? extends Modification> modificationType) {
new JavaTestKit(getSystem()) {{
- new Within(duration("1 seconds")) {
+ new Within(duration("3 seconds")) {
@Override
protected void run() {
subject
- .tell(new ShardTransaction.GetCompositedModification(),
+ .tell(new ShardWriteTransaction.GetCompositedModification(),
getRef());
final CompositeModification compositeModification =
- new ExpectMsg<CompositeModification>(duration("1 seconds"), "match hint") {
+ new ExpectMsg<CompositeModification>(duration("3 seconds"), "match hint") {
// do not put code outside this method, will run afterwards
@Override
protected CompositeModification match(Object in) {
- if (in instanceof ShardTransaction.GetCompositeModificationReply) {
- return ((ShardTransaction.GetCompositeModificationReply) in)
+ if (in instanceof ShardWriteTransaction.GetCompositeModificationReply) {
+ return ((ShardWriteTransaction.GetCompositeModificationReply) in)
.getModification();
} else {
throw noMatch();
@Test
public void testOnReceiveWriteData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props(store.newWriteOnlyTransaction(), shard, TestModel.createTestContext());
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newWriteOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject =
getSystem().actorOf(props, "testWriteData");
@Test
public void testOnReceiveMergeData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props(store.newReadWriteTransaction(), shard, testSchemaContext);
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject =
getSystem().actorOf(props, "testMergeData");
@Test
public void testOnReceiveDeleteData() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props( store.newWriteOnlyTransaction(), shard, TestModel.createTestContext());
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props( store.newWriteOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject =
getSystem().actorOf(props, "testDeleteData");
@Test
public void testOnReceiveReadyTransaction() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props( store.newReadWriteTransaction(), shard, TestModel.createTestContext());
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props( store.newReadWriteTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject =
getSystem().actorOf(props, "testReadyTransaction");
@Test
public void testOnReceiveCloseTransaction() throws Exception {
new JavaTestKit(getSystem()) {{
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props(store.newReadWriteTransaction(), shard, TestModel.createTestContext());
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
final ActorRef subject =
getSystem().actorOf(props, "testCloseTransaction");
watch(subject);
- new Within(duration("2 seconds")) {
+ new Within(duration("6 seconds")) {
@Override
protected void run() {
subject.tell(new CloseTransaction().toSerializable(), getRef());
- final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
+ final String out = new ExpectMsg<String>(duration("3 seconds"), "match hint") {
// do not put code outside this method, will run afterwards
@Override
protected String match(Object in) {
+ System.out.println("!!!IN match 1: "+(in!=null?in.getClass():"NULL"));
if (in.getClass().equals(CloseTransactionReply.SERIALIZABLE_CLASS)) {
return "match";
} else {
assertEquals("match", out);
- final String termination = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
+ final String termination = new ExpectMsg<String>(duration("3 seconds"), "match hint") {
// do not put code outside this method, will run afterwards
@Override
protected String match(Object in) {
+ System.out.println("!!!IN match 2: "+(in!=null?in.getClass():"NULL"));
if (in instanceof Terminated) {
return "match";
} else {
}
}.get(); // this extracts the received message
-
- expectNoMsg();
+ assertEquals("match", termination);
}
-
-
};
}};
+ }
+ @Test(expected=UnknownMessageException.class)
+ public void testNegativePerformingWriteOperationOnReadTransaction() throws Exception {
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
+ final TestActorRef subject = TestActorRef.apply(props,getSystem());
+
+ subject.receive(new DeleteData(TestModel.TEST_PATH).toSerializable(), ActorRef.noSender());
}
+ @Test
+ public void testShardTransactionInactivity() {
- @Test
- public void testNegativePerformingWriteOperationOnReadTransaction() throws Exception {
- try {
+ datastoreContext = DatastoreContext.newBuilder().shardTransactionIdleTimeout(
+ Duration.create(500, TimeUnit.MILLISECONDS)).build();
- final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER, Collections.EMPTY_MAP, null));
- final Props props =
- ShardTransaction.props(store.newReadOnlyTransaction(), shard, TestModel.createTestContext());
- final TestActorRef subject = TestActorRef.apply(props,getSystem());
+ new JavaTestKit(getSystem()) {{
+ final ActorRef shard = createShard();
+ final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
+ testSchemaContext, datastoreContext, shardStats, "txn");
+ final ActorRef subject =
+ getSystem().actorOf(props, "testShardTransactionInactivity");
- subject.receive(new DeleteData(TestModel.TEST_PATH).toSerializable(), ActorRef.noSender());
- Assert.assertFalse(true);
+ watch(subject);
+ // The shard Tx actor should receive a ReceiveTimeout message and self-destruct.
- } catch (Exception cs) {
- assertEquals(UnknownMessageException.class.getSimpleName(), cs.getClass().getSimpleName());
- assertTrue(cs.getMessage(), cs.getMessage().startsWith("Unknown message received "));
+ final String termination = new ExpectMsg<String>(duration("3 seconds"), "match hint") {
+ // do not put code outside this method, will run afterwards
+ @Override
+ protected String match(Object in) {
+ if (in instanceof Terminated) {
+ return "match";
+ } else {
+ throw noMatch();
+ }
+ }
+ }.get(); // this extracts the received message
+
+ assertEquals("match", termination);
+ }};
}
- }
}
Code Review / controller.git / blobdiff