1 package org.opendaylight.controller.cluster.datastore;
3 import akka.actor.ActorRef;
4 import akka.actor.Props;
5 import akka.actor.Terminated;
6 import akka.testkit.JavaTestKit;
7 import akka.testkit.TestActorRef;
9 import com.google.common.util.concurrent.ListeningExecutorService;
10 import com.google.common.util.concurrent.MoreExecutors;
12 import org.junit.BeforeClass;
13 import org.junit.Test;
14 import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
15 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
16 import org.opendaylight.controller.cluster.datastore.messages.CloseTransaction;
17 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionReply;
18 import org.opendaylight.controller.cluster.datastore.messages.DataExists;
19 import org.opendaylight.controller.cluster.datastore.messages.DataExistsReply;
20 import org.opendaylight.controller.cluster.datastore.messages.DeleteData;
21 import org.opendaylight.controller.cluster.datastore.messages.DeleteDataReply;
22 import org.opendaylight.controller.cluster.datastore.messages.MergeData;
23 import org.opendaylight.controller.cluster.datastore.messages.MergeDataReply;
24 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
25 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
26 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransaction;
27 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
28 import org.opendaylight.controller.cluster.datastore.messages.WriteData;
29 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
30 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
31 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
32 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
33 import org.opendaylight.controller.cluster.datastore.modification.Modification;
34 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
35 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
36 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
37 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties;
38 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
39 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
40 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
42 import scala.concurrent.duration.Duration;
44 import java.util.Collections;
45 import java.util.concurrent.TimeUnit;
47 import static org.junit.Assert.assertEquals;
48 import static org.junit.Assert.assertTrue;
50 public class ShardTransactionTest extends AbstractActorTest {
51 private static ListeningExecutorService storeExecutor =
52 MoreExecutors.listeningDecorator(MoreExecutors.sameThreadExecutor());
54 private static final InMemoryDOMDataStore store =
55 new InMemoryDOMDataStore("OPER", storeExecutor, MoreExecutors.sameThreadExecutor());
57 private static final SchemaContext testSchemaContext = TestModel.createTestContext();
59 private static final ShardIdentifier SHARD_IDENTIFIER =
60 ShardIdentifier.builder().memberName("member-1")
61 .shardName("inventory").type("config").build();
63 private DatastoreContext datastoreContext = new DatastoreContext();
66 public static void staticSetup() {
67 store.onGlobalContextUpdated(testSchemaContext);
71 public void testOnReceiveReadData() throws Exception {
72 new JavaTestKit(getSystem()) {{
73 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
74 Collections.EMPTY_MAP, new DatastoreContext()));
75 final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
76 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
77 final ActorRef subject = getSystem().actorOf(props, "testReadData");
79 new Within(duration("1 seconds")) {
81 protected void run() {
84 new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(),
87 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
88 // do not put code outside this method, will run afterwards
90 protected String match(Object in) {
91 if (in.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
92 if (ReadDataReply.fromSerializable(testSchemaContext,YangInstanceIdentifier.builder().build(), in)
93 .getNormalizedNode()!= null) {
101 }.get(); // this extracts the received message
103 assertEquals("match", out);
114 public void testOnReceiveReadDataWhenDataNotFound() throws Exception {
115 new JavaTestKit(getSystem()) {{
116 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
117 Collections.EMPTY_MAP, new DatastoreContext()));
118 final Props props = ShardTransaction.props( store.newReadOnlyTransaction(), shard,
119 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
120 final ActorRef subject = getSystem().actorOf(props, "testReadDataWhenDataNotFound");
122 new Within(duration("1 seconds")) {
124 protected void run() {
127 new ReadData(TestModel.TEST_PATH).toSerializable(),
130 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
131 // do not put code outside this method, will run afterwards
133 protected String match(Object in) {
134 if (in.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
135 if (ReadDataReply.fromSerializable(testSchemaContext,TestModel.TEST_PATH, in)
145 }.get(); // this extracts the received message
147 assertEquals("match", out);
158 public void testOnReceiveDataExistsPositive() throws Exception {
159 new JavaTestKit(getSystem()) {{
160 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
161 Collections.EMPTY_MAP, new DatastoreContext()));
162 final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
163 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
164 final ActorRef subject = getSystem().actorOf(props, "testDataExistsPositive");
166 new Within(duration("1 seconds")) {
168 protected void run() {
171 new DataExists(YangInstanceIdentifier.builder().build()).toSerializable(),
174 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
175 // do not put code outside this method, will run afterwards
177 protected String match(Object in) {
178 if (in.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) {
179 if (DataExistsReply.fromSerializable(in)
188 }.get(); // this extracts the received message
190 assertEquals("match", out);
201 public void testOnReceiveDataExistsNegative() throws Exception {
202 new JavaTestKit(getSystem()) {{
203 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
204 Collections.EMPTY_MAP, new DatastoreContext()));
205 final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
206 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
207 final ActorRef subject = getSystem().actorOf(props, "testDataExistsNegative");
209 new Within(duration("1 seconds")) {
211 protected void run() {
214 new DataExists(TestModel.TEST_PATH).toSerializable(),
217 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
218 // do not put code outside this method, will run afterwards
220 protected String match(Object in) {
221 if (in.getClass().equals(DataExistsReply.SERIALIZABLE_CLASS)) {
222 if (!DataExistsReply.fromSerializable(in)
231 }.get(); // this extracts the received message
233 assertEquals("match", out);
243 private void assertModification(final ActorRef subject,
244 final Class<? extends Modification> modificationType) {
245 new JavaTestKit(getSystem()) {{
246 new Within(duration("1 seconds")) {
248 protected void run() {
250 .tell(new ShardTransaction.GetCompositedModification(),
253 final CompositeModification compositeModification =
254 new ExpectMsg<CompositeModification>(duration("1 seconds"), "match hint") {
255 // do not put code outside this method, will run afterwards
257 protected CompositeModification match(Object in) {
258 if (in instanceof ShardTransaction.GetCompositeModificationReply) {
259 return ((ShardTransaction.GetCompositeModificationReply) in)
265 }.get(); // this extracts the received message
268 compositeModification.getModifications().size() == 1);
269 assertEquals(modificationType,
270 compositeModification.getModifications().get(0)
279 public void testOnReceiveWriteData() throws Exception {
280 new JavaTestKit(getSystem()) {{
281 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
282 Collections.EMPTY_MAP, new DatastoreContext()));
283 final Props props = ShardTransaction.props(store.newWriteOnlyTransaction(), shard,
284 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
285 final ActorRef subject =
286 getSystem().actorOf(props, "testWriteData");
288 new Within(duration("1 seconds")) {
290 protected void run() {
292 subject.tell(new WriteData(TestModel.TEST_PATH,
293 ImmutableNodes.containerNode(TestModel.TEST_QNAME), TestModel.createTestContext()).toSerializable(),
296 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
297 // do not put code outside this method, will run afterwards
299 protected String match(Object in) {
300 if (in.getClass().equals(WriteDataReply.SERIALIZABLE_CLASS)) {
306 }.get(); // this extracts the received message
308 assertEquals("match", out);
310 assertModification(subject, WriteModification.class);
320 public void testOnReceiveMergeData() throws Exception {
321 new JavaTestKit(getSystem()) {{
322 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
323 Collections.EMPTY_MAP, new DatastoreContext()));
324 final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
325 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
326 final ActorRef subject =
327 getSystem().actorOf(props, "testMergeData");
329 new Within(duration("1 seconds")) {
331 protected void run() {
333 subject.tell(new MergeData(TestModel.TEST_PATH,
334 ImmutableNodes.containerNode(TestModel.TEST_QNAME), testSchemaContext).toSerializable(),
337 final String out = new ExpectMsg<String>(duration("500 milliseconds"), "match hint") {
338 // do not put code outside this method, will run afterwards
340 protected String match(Object in) {
341 if (in.getClass().equals(MergeDataReply.SERIALIZABLE_CLASS)) {
347 }.get(); // this extracts the received message
349 assertEquals("match", out);
351 assertModification(subject, MergeModification.class);
362 public void testOnReceiveDeleteData() throws Exception {
363 new JavaTestKit(getSystem()) {{
364 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
365 Collections.EMPTY_MAP, new DatastoreContext()));
366 final Props props = ShardTransaction.props( store.newWriteOnlyTransaction(), shard,
367 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
368 final ActorRef subject =
369 getSystem().actorOf(props, "testDeleteData");
371 new Within(duration("1 seconds")) {
373 protected void run() {
375 subject.tell(new DeleteData(TestModel.TEST_PATH).toSerializable(), getRef());
377 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
378 // do not put code outside this method, will run afterwards
380 protected String match(Object in) {
381 if (in.getClass().equals(DeleteDataReply.SERIALIZABLE_CLASS)) {
387 }.get(); // this extracts the received message
389 assertEquals("match", out);
391 assertModification(subject, DeleteModification.class);
402 public void testOnReceiveReadyTransaction() throws Exception {
403 new JavaTestKit(getSystem()) {{
404 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
405 Collections.EMPTY_MAP, new DatastoreContext()));
406 final Props props = ShardTransaction.props( store.newReadWriteTransaction(), shard,
407 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
408 final ActorRef subject =
409 getSystem().actorOf(props, "testReadyTransaction");
411 new Within(duration("1 seconds")) {
413 protected void run() {
415 subject.tell(new ReadyTransaction().toSerializable(), getRef());
417 final String out = new ExpectMsg<String>(duration("1 seconds"), "match hint") {
418 // do not put code outside this method, will run afterwards
420 protected String match(Object in) {
421 if (in.getClass().equals(ReadyTransactionReply.SERIALIZABLE_CLASS)) {
427 }.get(); // this extracts the received message
429 assertEquals("match", out);
441 public void testOnReceiveCloseTransaction() throws Exception {
442 new JavaTestKit(getSystem()) {{
443 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
444 Collections.EMPTY_MAP, new DatastoreContext()));
445 final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
446 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
447 final ActorRef subject =
448 getSystem().actorOf(props, "testCloseTransaction");
452 new Within(duration("6 seconds")) {
454 protected void run() {
456 subject.tell(new CloseTransaction().toSerializable(), getRef());
458 final String out = new ExpectMsg<String>(duration("3 seconds"), "match hint") {
459 // do not put code outside this method, will run afterwards
461 protected String match(Object in) {
462 System.out.println("!!!IN match 1: "+(in!=null?in.getClass():"NULL"));
463 if (in.getClass().equals(CloseTransactionReply.SERIALIZABLE_CLASS)) {
469 }.get(); // this extracts the received message
471 assertEquals("match", out);
473 final String termination = new ExpectMsg<String>(duration("3 seconds"), "match hint") {
474 // do not put code outside this method, will run afterwards
476 protected String match(Object in) {
477 System.out.println("!!!IN match 2: "+(in!=null?in.getClass():"NULL"));
478 if (in instanceof Terminated) {
484 }.get(); // this extracts the received message
486 assertEquals("match", termination);
492 @Test(expected=UnknownMessageException.class)
493 public void testNegativePerformingWriteOperationOnReadTransaction() throws Exception {
494 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
495 Collections.EMPTY_MAP, new DatastoreContext()));
496 final Props props = ShardTransaction.props(store.newReadOnlyTransaction(), shard,
497 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
498 final TestActorRef subject = TestActorRef.apply(props,getSystem());
500 subject.receive(new DeleteData(TestModel.TEST_PATH).toSerializable(), ActorRef.noSender());
504 public void testShardTransactionInactivity() {
506 datastoreContext = new DatastoreContext(InMemoryDOMDataStoreConfigProperties.getDefault(),
507 Duration.create(500, TimeUnit.MILLISECONDS));
509 new JavaTestKit(getSystem()) {{
510 final ActorRef shard = getSystem().actorOf(Shard.props(SHARD_IDENTIFIER,
511 Collections.EMPTY_MAP, new DatastoreContext()));
512 final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard,
513 testSchemaContext, datastoreContext,SHARD_IDENTIFIER.toString());
514 final ActorRef subject =
515 getSystem().actorOf(props, "testShardTransactionInactivity");
519 // The shard Tx actor should receive a ReceiveTimeout message and self-destruct.
521 final String termination = new ExpectMsg<String>(duration("3 seconds"), "match hint") {
522 // do not put code outside this method, will run afterwards
524 protected String match(Object in) {
525 if (in instanceof Terminated) {
531 }.get(); // this extracts the received message
533 assertEquals("match", termination);