opendaylight/md-sal/sal-remoterpc-connector/src/test/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistryTest.java

   1 package org.opendaylight.controller.remote.rpc.registry;
   2
   3 import akka.actor.ActorPath;
   4 import akka.actor.ActorRef;
   5 import akka.actor.ActorSelection;
   6 import akka.actor.ActorSystem;
   7 import akka.actor.ChildActorPath;
   8 import akka.actor.Props;
   9 import akka.japi.Pair;
  10 import akka.testkit.JavaTestKit;
  11 import com.typesafe.config.ConfigFactory;
  12 import org.junit.After;
  13 import org.junit.AfterClass;
  14 import org.junit.Assert;
  15 import org.junit.Before;
  16 import org.junit.BeforeClass;
  17 import org.junit.Test;
  18 import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
  19 import org.opendaylight.controller.sal.connector.api.RpcRouter;
  20 import org.opendaylight.yangtools.yang.common.QName;
  21 import scala.concurrent.Await;
  22 import scala.concurrent.Future;
  23 import scala.concurrent.duration.FiniteDuration;
  24
  25 import java.net.URI;
  26 import java.net.URISyntaxException;
  27 import java.util.ArrayList;
  28 import java.util.List;
  29 import java.util.concurrent.TimeUnit;
  30
  31 import static org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes;
  32 import static org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRouters;
  33 import static org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRoutersReply;
  34 import static org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter;
  35
  36 public class RpcRegistryTest {
  37
  38     private static ActorSystem node1;
  39     private static ActorSystem node2;
  40     private static ActorSystem node3;
  41
  42     private ActorRef registry1;
  43     private ActorRef registry2;
  44     private ActorRef registry3;
  45
  46     @BeforeClass
  47     public static void setup() throws InterruptedException {
  48         Thread.sleep(1000); //give some time for previous test to close netty ports
  49         node1 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberA"));
  50         node2 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberB"));
  51         node3 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberC"));
  52     }
  53
  54     @AfterClass
  55     public static void teardown(){
  56         JavaTestKit.shutdownActorSystem(node1);
  57         JavaTestKit.shutdownActorSystem(node2);
  58         JavaTestKit.shutdownActorSystem(node3);
  59         if (node1 != null)
  60             node1.shutdown();
  61         if (node2 != null)
  62             node2.shutdown();
  63         if (node3 != null)
  64             node3.shutdown();
  65
  66     }
  67
  68     @Before
  69     public void createRpcRegistry() throws InterruptedException {
  70         registry1 = node1.actorOf(Props.create(RpcRegistry.class));
  71         registry2 = node2.actorOf(Props.create(RpcRegistry.class));
  72         registry3 = node3.actorOf(Props.create(RpcRegistry.class));
  73     }
  74
  75     @After
  76     public void stopRpcRegistry() throws InterruptedException {
  77         if (registry1 != null)
  78             node1.stop(registry1);
  79         if (registry2 != null)
  80             node2.stop(registry2);
  81         if (registry3 != null)
  82             node3.stop(registry3);
  83     }
  84
  85     /**
  86      * One node cluster.
  87      * Register rpc. Ensure router can be found
  88      *
  89      * @throws URISyntaxException
  90      * @throws InterruptedException
  91      */
  92     @Test
  93     public void testWhenRpcAddedOneNodeShouldAppearOnSameNode() throws URISyntaxException, InterruptedException {
  94
  95         final JavaTestKit mockBroker = new JavaTestKit(node1);
  96
  97         //Add rpc on node 1
  98         registry1.tell(new SetLocalRouter(mockBroker.getRef()), mockBroker.getRef());
  99         registry1.tell(getAddRouteMessage(), mockBroker.getRef());
 100
 101         Thread.sleep(1000);//
 102
 103         //find the route on node 1's registry
 104         registry1.tell(new FindRouters(createRouteId()), mockBroker.getRef());
 105         FindRoutersReply message = mockBroker.expectMsgClass(JavaTestKit.duration("10 second"), FindRoutersReply.class);
 106         List<Pair<ActorRef, Long>> pairs = message.getRouterWithUpdateTime();
 107
 108         validateRouterReceived(pairs, mockBroker.getRef());
 109     }
 110
 111     /**
 112      * Three node cluster.
 113      * Register rpc on 1 node. Ensure its router can be found on other 2.
 114      *
 115      * @throws URISyntaxException
 116      * @throws InterruptedException
 117      */
 118     @Test
 119     public void testWhenRpcAddedOneNodeShouldAppearOnAnother() throws URISyntaxException, InterruptedException {
 120
 121         validateSystemStartup();
 122
 123         final JavaTestKit mockBroker1 = new JavaTestKit(node1);
 124         final JavaTestKit mockBroker2 = new JavaTestKit(node2);
 125         final JavaTestKit mockBroker3 = new JavaTestKit(node3);
 126
 127         //Add rpc on node 1
 128         registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
 129         registry1.tell(getAddRouteMessage(), mockBroker1.getRef());
 130
 131         Thread.sleep(5000);// give some time for bucket store data sync
 132
 133         //find the route in node 2's registry
 134         registry2.tell(new FindRouters(createRouteId()), mockBroker2.getRef());
 135         FindRoutersReply message = mockBroker2.expectMsgClass(JavaTestKit.duration("10 second"), FindRoutersReply.class);
 136         List<Pair<ActorRef, Long>> pairs = message.getRouterWithUpdateTime();
 137
 138         validateRouterReceived(pairs, mockBroker1.getRef());
 139
 140         //find the route in node 3's registry
 141         registry3.tell(new FindRouters(createRouteId()), mockBroker3.getRef());
 142         message = mockBroker3.expectMsgClass(JavaTestKit.duration("10 second"), FindRoutersReply.class);
 143         pairs = message.getRouterWithUpdateTime();
 144
 145         validateRouterReceived(pairs, mockBroker1.getRef());
 146
 147     }
 148
 149     /**
 150      * Three node cluster.
 151      * Register rpc on 2 nodes. Ensure 2 routers are found on 3rd.
 152      *
 153      * @throws Exception
 154      */
 155     @Test
 156     public void testAnRpcAddedOnMultiNodesShouldReturnMultiRouter() throws Exception {
 157
 158         validateSystemStartup();
 159
 160         final JavaTestKit mockBroker1 = new JavaTestKit(node1);
 161         final JavaTestKit mockBroker2 = new JavaTestKit(node2);
 162         final JavaTestKit mockBroker3 = new JavaTestKit(node3);
 163
 164         //Thread.sleep(5000);//let system come up
 165
 166         //Add rpc on node 1
 167         registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
 168         registry1.tell(getAddRouteMessage(), mockBroker1.getRef());
 169
 170         //Add same rpc on node 2
 171         registry2.tell(new SetLocalRouter(mockBroker2.getRef()), mockBroker2.getRef());
 172         registry2.tell(getAddRouteMessage(), mockBroker2.getRef());
 173
 174         registry3.tell(new SetLocalRouter(mockBroker3.getRef()), mockBroker3.getRef());
 175         Thread.sleep(5000);// give some time for bucket store data sync
 176
 177         //find the route in node 3's registry
 178         registry3.tell(new FindRouters(createRouteId()), mockBroker3.getRef());
 179         FindRoutersReply message = mockBroker3.expectMsgClass(JavaTestKit.duration("10 second"), FindRoutersReply.class);
 180         List<Pair<ActorRef, Long>> pairs = message.getRouterWithUpdateTime();
 181
 182         validateMultiRouterReceived(pairs, mockBroker1.getRef(), mockBroker2.getRef());
 183
 184     }
 185
 186     private void validateMultiRouterReceived(List<Pair<ActorRef, Long>> actual, ActorRef... expected) {
 187         Assert.assertTrue(actual != null);
 188         Assert.assertTrue(actual.size() == expected.length);
 189     }
 190
 191     private void validateRouterReceived(List<Pair<ActorRef, Long>> actual, ActorRef expected){
 192         Assert.assertTrue(actual != null);
 193         Assert.assertTrue(actual.size() == 1);
 194
 195         for (Pair<ActorRef, Long> pair : actual){
 196             Assert.assertTrue(expected.path().uid() == pair.first().path().uid());
 197         }
 198     }
 199
 200     private void validateSystemStartup() throws InterruptedException {
 201
 202         Thread.sleep(5000);
 203         ActorPath gossiper1Path = new ChildActorPath(new ChildActorPath(registry1.path(), "store"), "gossiper");
 204         ActorPath gossiper2Path = new ChildActorPath(new ChildActorPath(registry2.path(), "store"), "gossiper");
 205         ActorPath gossiper3Path = new ChildActorPath(new ChildActorPath(registry3.path(), "store"), "gossiper");
 206
 207         ActorSelection gossiper1 = node1.actorSelection(gossiper1Path);
 208         ActorSelection gossiper2 = node2.actorSelection(gossiper2Path);
 209         ActorSelection gossiper3 = node3.actorSelection(gossiper3Path);
 210
 211
 212         if (!resolveReference(gossiper1, gossiper2, gossiper3))
 213             Assert.fail("Could not find gossipers");
 214     }
 215
 216     private Boolean resolveReference(ActorSelection... gossipers) throws InterruptedException {
 217
 218         Boolean resolved = true;
 219
 220         for (int i=0; i< 5; i++) {
 221             Thread.sleep(1000);
 222             for (ActorSelection gossiper : gossipers) {
 223                 Future<ActorRef> future = gossiper.resolveOne(new FiniteDuration(5000, TimeUnit.MILLISECONDS));
 224
 225                 ActorRef ref = null;
 226                 try {
 227                     ref = Await.result(future, new FiniteDuration(10000, TimeUnit.MILLISECONDS));
 228                 } catch (Exception e) {
 229                     e.printStackTrace();
 230                 }
 231
 232                 if (ref == null)
 233                     resolved = false;
 234             }
 235
 236             if (resolved) break;
 237         }
 238         return resolved;
 239     }
 240
 241     private AddOrUpdateRoutes getAddRouteMessage() throws URISyntaxException {
 242         return new AddOrUpdateRoutes(createRouteIds());
 243     }
 244
 245     private List<RpcRouter.RouteIdentifier<?,?,?>> createRouteIds() throws URISyntaxException {
 246         QName type = new QName(new URI("/mockrpc"), "mockrpc");
 247         List<RpcRouter.RouteIdentifier<?,?,?>> routeIds = new ArrayList<>();
 248         routeIds.add(new RouteIdentifierImpl(null, type, null));
 249         return routeIds;
 250     }
 251
 252     private RpcRouter.RouteIdentifier<?,?,?> createRouteId() throws URISyntaxException {
 253         QName type = new QName(new URI("/mockrpc"), "mockrpc");
 254         return new RouteIdentifierImpl(null, type, null);
 255     }
 256 }