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.pattern.Patterns;
  10 import akka.testkit.JavaTestKit;
  11 import akka.util.Timeout;
  12 import com.google.common.base.Predicate;
  13 import java.net.URI;
  14 import java.net.URISyntaxException;
  15 import java.util.ArrayList;
  16 import java.util.List;
  17 import java.util.concurrent.TimeUnit;
  18 import java.util.concurrent.TimeoutException;
  19 import javax.annotation.Nullable;
  20 import org.junit.After;
  21 import org.junit.AfterClass;
  22 import org.junit.Before;
  23 import org.junit.BeforeClass;
  24 import org.junit.Test;
  25 import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
  26 import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
  27 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes;
  28 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.RemoveRoutes;
  29 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter;
  30 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages;
  31 import org.opendaylight.controller.sal.connector.api.RpcRouter;
  32 import org.opendaylight.controller.utils.ConditionalProbe;
  33 import org.opendaylight.yangtools.yang.common.QName;
  34 import scala.concurrent.Await;
  35 import scala.concurrent.Future;
  36 import scala.concurrent.duration.Duration;
  37 import scala.concurrent.duration.FiniteDuration;
  38
  39 public class RpcRegistryTest {
  40
  41     private static ActorSystem node1;
  42     private static ActorSystem node2;
  43     private static ActorSystem node3;
  44
  45     private ActorRef registry1;
  46     private ActorRef registry2;
  47     private ActorRef registry3;
  48
  49     @BeforeClass
  50     public static void staticSetup() throws InterruptedException {
  51       RemoteRpcProviderConfig config1 = new RemoteRpcProviderConfig.Builder("memberA").build();
  52       RemoteRpcProviderConfig config2 = new RemoteRpcProviderConfig.Builder("memberB").build();
  53       RemoteRpcProviderConfig config3 = new RemoteRpcProviderConfig.Builder("memberC").build();
  54       node1 = ActorSystem.create("opendaylight-rpc", config1.get());
  55       node2 = ActorSystem.create("opendaylight-rpc", config2.get());
  56       node3 = ActorSystem.create("opendaylight-rpc", config3.get());
  57     }
  58
  59     @AfterClass
  60     public static void staticTeardown() {
  61       JavaTestKit.shutdownActorSystem(node1);
  62       JavaTestKit.shutdownActorSystem(node2);
  63       JavaTestKit.shutdownActorSystem(node3);
  64     }
  65
  66     @Before
  67     public void setup() {
  68         registry1 = node1.actorOf(Props.create(RpcRegistry.class));
  69         registry2 = node2.actorOf(Props.create(RpcRegistry.class));
  70         registry3 = node3.actorOf(Props.create(RpcRegistry.class));
  71     }
  72
  73     @After
  74     public void teardown() {
  75         if (registry1 != null) {
  76             node1.stop(registry1);
  77         }
  78         if (registry2 != null) {
  79             node2.stop(registry2);
  80         }
  81         if (registry3 != null) {
  82             node3.stop(registry3);
  83         }
  84     }
  85
  86     /**
  87      * One node cluster. 1. Register rpc, ensure router can be found 2. Then remove rpc, ensure its
  88      * deleted
  89      *
  90      * @throws URISyntaxException
  91      * @throws InterruptedException
  92      */
  93     @Test
  94     public void testAddRemoveRpcOnSameNode() throws Exception {
  95
  96         System.out.println("testAddRemoveRpcOnSameNode starting");
  97
  98         final JavaTestKit mockBroker = new JavaTestKit(node1);
  99
 100         final ActorPath bucketStorePath = new ChildActorPath(registry1.path(), "store");
 101
 102         // Add rpc on node 1
 103         registry1.tell(new SetLocalRouter(mockBroker.getRef()), mockBroker.getRef());
 104
 105         // install probe
 106         final JavaTestKit probe1 = createProbeForMessage(node1, bucketStorePath,
 107                 Messages.BucketStoreMessages.UpdateBucket.class);
 108
 109         registry1.tell(getAddRouteMessage(), mockBroker.getRef());
 110
 111         // Bucket store should get an update bucket message. Updated bucket contains added rpc.
 112         probe1.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS),
 113                 Messages.BucketStoreMessages.UpdateBucket.class);
 114
 115         // Now remove rpc
 116         registry1.tell(getRemoveRouteMessage(), mockBroker.getRef());
 117
 118         // Bucket store should get an update bucket message. Rpc is removed in the updated bucket
 119         probe1.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS),
 120                 Messages.BucketStoreMessages.UpdateBucket.class);
 121
 122         System.out.println("testAddRemoveRpcOnSameNode ending");
 123
 124     }
 125
 126     /**
 127      * Three node cluster. 1. Register rpc on 1 node, ensure 2nd node gets updated 2. Remove rpc on
 128      * 1 node, ensure 2nd node gets updated
 129      *
 130      * @throws URISyntaxException
 131      * @throws InterruptedException
 132      */
 133     @Test
 134     public void testRpcAddRemoveInCluster() throws Exception {
 135
 136         System.out.println("testRpcAddRemoveInCluster starting");
 137
 138         final JavaTestKit mockBroker1 = new JavaTestKit(node1);
 139
 140         // install probe on node2's bucket store
 141         final ActorPath bucketStorePath = new ChildActorPath(registry2.path(), "store");
 142         final JavaTestKit probe2 = createProbeForMessage(node2, bucketStorePath,
 143                 Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
 144
 145         // Add rpc on node 1
 146         registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
 147         registry1.tell(getAddRouteMessage(), mockBroker1.getRef());
 148
 149         // Bucket store on node2 should get a message to update its local copy of remote buckets
 150         probe2.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS),
 151                 Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
 152
 153         // Now remove
 154         registry1.tell(getRemoveRouteMessage(), mockBroker1.getRef());
 155
 156         // Bucket store on node2 should get a message to update its local copy of remote buckets
 157         probe2.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS),
 158                 Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
 159
 160         System.out.println("testRpcAddRemoveInCluster ending");
 161     }
 162
 163     /**
 164      * Three node cluster. Register rpc on 2 nodes. Ensure 3rd gets updated.
 165      *
 166      * @throws Exception
 167      */
 168     @Test
 169     public void testRpcAddedOnMultiNodes() throws Exception {
 170
 171         final JavaTestKit mockBroker1 = new JavaTestKit(node1);
 172         final JavaTestKit mockBroker2 = new JavaTestKit(node2);
 173         final JavaTestKit mockBroker3 = new JavaTestKit(node3);
 174
 175         registry3.tell(new SetLocalRouter(mockBroker3.getRef()), mockBroker3.getRef());
 176
 177         // install probe on node 3
 178         final ActorPath bucketStorePath = new ChildActorPath(registry3.path(), "store");
 179         final JavaTestKit probe3 = createProbeForMessage(node3, bucketStorePath,
 180                 Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
 181
 182         // Add rpc on node 1
 183         registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
 184         registry1.tell(getAddRouteMessage(), mockBroker1.getRef());
 185
 186         probe3.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS),
 187                 Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
 188
 189         // Add same rpc on node 2
 190         registry2.tell(new SetLocalRouter(mockBroker2.getRef()), mockBroker2.getRef());
 191         registry2.tell(getAddRouteMessage(), mockBroker2.getRef());
 192
 193         probe3.expectMsgClass(FiniteDuration.apply(10, TimeUnit.SECONDS),
 194                 Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
 195     }
 196
 197     private JavaTestKit createProbeForMessage(ActorSystem node, ActorPath subjectPath, final Class<?> clazz)
 198             throws Exception {
 199         final JavaTestKit probe = new JavaTestKit(node);
 200
 201         ConditionalProbe conditionalProbe = new ConditionalProbe(probe.getRef(), new Predicate<Object>() {
 202             @Override
 203             public boolean apply(@Nullable Object input) {
 204                 if (input != null) {
 205                     return clazz.equals(input.getClass());
 206                 } else {
 207                     return false;
 208                 }
 209             }
 210         });
 211
 212         FiniteDuration duration = Duration.create(3, TimeUnit.SECONDS);
 213         Timeout timeout = new Timeout(duration);
 214         int maxTries = 30;
 215         int i = 0;
 216         while(true) {
 217             ActorSelection subject = node.actorSelection(subjectPath);
 218             Future<Object> future = Patterns.ask(subject, conditionalProbe, timeout);
 219
 220             try {
 221                 Await.ready(future, duration);
 222                 break;
 223             } catch (TimeoutException | InterruptedException e) {
 224                 if(++i > maxTries) {
 225                     throw e;
 226                 }
 227             }
 228         }
 229
 230         return probe;
 231
 232     }
 233
 234     private AddOrUpdateRoutes getAddRouteMessage() throws URISyntaxException {
 235         return new AddOrUpdateRoutes(createRouteIds());
 236     }
 237
 238     private RemoveRoutes getRemoveRouteMessage() throws URISyntaxException {
 239         return new RemoveRoutes(createRouteIds());
 240     }
 241
 242     private List<RpcRouter.RouteIdentifier<?, ?, ?>> createRouteIds() throws URISyntaxException {
 243         QName type = new QName(new URI("/mockrpc"), "mockrpc");
 244         List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new ArrayList<>();
 245         routeIds.add(new RouteIdentifierImpl(null, type, null));
 246         return routeIds;
 247     }
 248
 249 }