opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/ActorContextTest.java

   1 package org.opendaylight.controller.cluster.datastore.utils;
   2
   3 import akka.actor.ActorRef;
   4 import akka.actor.ActorSelection;
   5 import akka.actor.ActorSystem;
   6 import akka.actor.Address;
   7 import akka.actor.Props;
   8 import akka.actor.UntypedActor;
   9 import akka.japi.Creator;
  10 import akka.testkit.JavaTestKit;
  11 import com.google.common.base.Optional;
  12 import java.util.concurrent.TimeUnit;
  13 import org.junit.Test;
  14 import org.opendaylight.controller.cluster.datastore.AbstractActorTest;
  15 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
  16 import org.opendaylight.controller.cluster.datastore.Configuration;
  17 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  18 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  19 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
  20 import scala.concurrent.Await;
  21 import scala.concurrent.Future;
  22 import scala.concurrent.duration.Duration;
  23
  24 import static org.junit.Assert.assertEquals;
  25 import static org.junit.Assert.assertTrue;
  26 import static org.mockito.Mockito.mock;
  27
  28 public class ActorContextTest extends AbstractActorTest{
  29
  30     private static class MockShardManager extends UntypedActor {
  31
  32         private final boolean found;
  33         private final ActorRef actorRef;
  34
  35         private MockShardManager(boolean found, ActorRef actorRef){
  36
  37             this.found = found;
  38             this.actorRef = actorRef;
  39         }
  40
  41         @Override public void onReceive(Object message) throws Exception {
  42             if(found){
  43                 getSender().tell(new LocalShardFound(actorRef), getSelf());
  44             } else {
  45                 getSender().tell(new LocalShardNotFound(((FindLocalShard) message).getShardName()), getSelf());
  46             }
  47         }
  48
  49         private static Props props(final boolean found, final ActorRef actorRef){
  50             return Props.create(new MockShardManagerCreator(found, actorRef) );
  51         }
  52
  53         @SuppressWarnings("serial")
  54         private static class MockShardManagerCreator implements Creator<MockShardManager> {
  55             final boolean found;
  56             final ActorRef actorRef;
  57
  58             MockShardManagerCreator(boolean found, ActorRef actorRef) {
  59                 this.found = found;
  60                 this.actorRef = actorRef;
  61             }
  62
  63             @Override
  64             public MockShardManager create() throws Exception {
  65                 return new MockShardManager(found, actorRef);
  66             }
  67         }
  68     }
  69
  70     @Test
  71     public void testFindLocalShardWithShardFound(){
  72         new JavaTestKit(getSystem()) {{
  73
  74             new Within(duration("1 seconds")) {
  75                 @Override
  76                 protected void run() {
  77
  78                     ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
  79
  80                     ActorRef shardManagerActorRef = getSystem()
  81                         .actorOf(MockShardManager.props(true, shardActorRef));
  82
  83                     ActorContext actorContext =
  84                         new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
  85                             mock(Configuration.class));
  86
  87                     Optional<ActorRef> out = actorContext.findLocalShard("default");
  88
  89                     assertEquals(shardActorRef, out.get());
  90
  91
  92                     expectNoMsg();
  93                 }
  94             };
  95         }};
  96
  97     }
  98
  99     @Test
 100     public void testFindLocalShardWithShardNotFound(){
 101         new JavaTestKit(getSystem()) {{
 102             ActorRef shardManagerActorRef = getSystem()
 103                     .actorOf(MockShardManager.props(false, null));
 104
 105             ActorContext actorContext =
 106                     new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
 107                             mock(Configuration.class));
 108
 109             Optional<ActorRef> out = actorContext.findLocalShard("default");
 110             assertTrue(!out.isPresent());
 111         }};
 112
 113     }
 114
 115     @Test
 116     public void testExecuteRemoteOperation() {
 117         new JavaTestKit(getSystem()) {{
 118             ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
 119
 120             ActorRef shardManagerActorRef = getSystem()
 121                     .actorOf(MockShardManager.props(true, shardActorRef));
 122
 123             ActorContext actorContext =
 124                     new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
 125                             mock(Configuration.class));
 126
 127             ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
 128
 129             Object out = actorContext.executeOperation(actor, "hello");
 130
 131             assertEquals("hello", out);
 132         }};
 133     }
 134
 135     @Test
 136     public void testExecuteRemoteOperationAsync() {
 137         new JavaTestKit(getSystem()) {{
 138             ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
 139
 140             ActorRef shardManagerActorRef = getSystem()
 141                     .actorOf(MockShardManager.props(true, shardActorRef));
 142
 143             ActorContext actorContext =
 144                     new ActorContext(getSystem(), shardManagerActorRef , mock(ClusterWrapper.class),
 145                             mock(Configuration.class));
 146
 147             ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
 148
 149             Future<Object> future = actorContext.executeOperationAsync(actor, "hello");
 150
 151             try {
 152                 Object result = Await.result(future, Duration.create(3, TimeUnit.SECONDS));
 153                 assertEquals("Result", "hello", result);
 154             } catch(Exception e) {
 155                 throw new AssertionError(e);
 156             }
 157         }};
 158     }
 159
 160     @Test
 161     public void testIsPathLocal() {
 162         MockClusterWrapper clusterWrapper = new MockClusterWrapper();
 163         ActorContext actorContext = null;
 164
 165         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 166         assertEquals(false, actorContext.isPathLocal(null));
 167         assertEquals(false, actorContext.isPathLocal(""));
 168
 169         clusterWrapper.setSelfAddress(null);
 170         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 171         assertEquals(false, actorContext.isPathLocal(""));
 172
 173         // even if the path is in local format, match the primary path (first 3 elements) and return true
 174         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 175         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 176         assertEquals(true, actorContext.isPathLocal("akka://test/user/$a"));
 177
 178         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 179         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 180         assertEquals(true, actorContext.isPathLocal("akka://test/user/$a"));
 181
 182         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 183         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 184         assertEquals(true, actorContext.isPathLocal("akka://test/user/token2/token3/$a"));
 185
 186         // self address of remote format,but Tx path local format.
 187         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 188         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 189         assertEquals(true, actorContext.isPathLocal(
 190             "akka://system/user/shardmanager/shard/transaction"));
 191
 192         // self address of local format,but Tx path remote format.
 193         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system"));
 194         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 195         assertEquals(false, actorContext.isPathLocal(
 196             "akka://system@127.0.0.1:2550/user/shardmanager/shard/transaction"));
 197
 198         //local path but not same
 199         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 200         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 201         assertEquals(true, actorContext.isPathLocal("akka://test1/user/$a"));
 202
 203         //ip and port same
 204         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 205         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 206         assertEquals(true, actorContext.isPathLocal("akka.tcp://system@127.0.0.1:2550/"));
 207
 208         // forward-slash missing in address
 209         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 210         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 211         assertEquals(false, actorContext.isPathLocal("akka.tcp://system@127.0.0.1:2550"));
 212
 213         //ips differ
 214         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 215         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 216         assertEquals(false, actorContext.isPathLocal("akka.tcp://system@127.1.0.1:2550/"));
 217
 218         //ports differ
 219         clusterWrapper.setSelfAddress(new Address("akka.tcp", "system", "127.0.0.1", 2550));
 220         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 221         assertEquals(false, actorContext.isPathLocal("akka.tcp://system@127.0.0.1:2551/"));
 222     }
 223
 224     @Test
 225     public void testResolvePathForRemoteActor() {
 226         ActorContext actorContext =
 227                 new ActorContext(mock(ActorSystem.class), mock(ActorRef.class), mock(
 228                         ClusterWrapper.class),
 229                         mock(Configuration.class));
 230
 231         String actual = actorContext.resolvePath(
 232                 "akka.tcp://system@127.0.0.1:2550/user/shardmanager/shard",
 233                 "akka://system/user/shardmanager/shard/transaction");
 234
 235         String expected = "akka.tcp://system@127.0.0.1:2550/user/shardmanager/shard/transaction";
 236
 237         assertEquals(expected, actual);
 238     }
 239
 240     @Test
 241     public void testResolvePathForLocalActor() {
 242         ActorContext actorContext =
 243                 new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
 244                         mock(Configuration.class));
 245
 246         String actual = actorContext.resolvePath(
 247                 "akka://system/user/shardmanager/shard",
 248                 "akka://system/user/shardmanager/shard/transaction");
 249
 250         String expected = "akka://system/user/shardmanager/shard/transaction";
 251
 252         assertEquals(expected, actual);
 253     }
 254
 255     @Test
 256     public void testResolvePathForRemoteActorWithProperRemoteAddress() {
 257         ActorContext actorContext =
 258                 new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
 259                         mock(Configuration.class));
 260
 261         String actual = actorContext.resolvePath(
 262                 "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard",
 263                 "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard/transaction");
 264
 265         String expected = "akka.tcp://system@7.0.0.1:2550/user/shardmanager/shard/transaction";
 266
 267         assertEquals(expected, actual);
 268     }
 269
 270 }