- public void testRoutedRpc() throws URISyntaxException {
- new JavaTestKit(system) {{
- ActorRef rpcRegistry = system.actorOf(RpcRegistry.props(Mockito.mock(ClusterWrapper.class)));
- QName type = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- final String route = "actor1";
-
- Set<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new HashSet<>();
- routeIds.add(routeId);
-
- AddRoutedRpc rpcMsg = new AddRoutedRpc(routeIds, route);
- rpcRegistry.tell(rpcMsg, getRef());
- expectMsgEquals(duration("2 second"), "Success");
-
- GetRoutedRpc getRpc = new GetRoutedRpc(routeId);
- rpcRegistry.tell(getRpc, getRef());
-
- Boolean getMsg = new ExpectMsg<Boolean>("GetRoutedRpcReply") {
- protected Boolean match(Object in) {
- if (in instanceof GetRoutedRpcReply) {
- GetRoutedRpcReply reply = (GetRoutedRpcReply)in;
- return route.equals(reply.getRoutePath());
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
+ public void testRpcAddRemoveInCluster() throws URISyntaxException, InterruptedException {
+
+ final JavaTestKit mockBroker1 = new JavaTestKit(node1);
+
+ //install probe on node2's bucket store
+ final ActorPath bucketStorePath = new ChildActorPath(registry2.path(), "store");
+ final JavaTestKit probe2 = createProbeForMessage(
+ node2, bucketStorePath, Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
+
+ //Add rpc on node 1
+ registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
+ registry1.tell(getAddRouteMessage(), mockBroker1.getRef());
+
+ //Bucket store on node2 should get a message to update its local copy of remote buckets
+ probe2.expectMsgClass(
+ FiniteDuration.apply(10, TimeUnit.SECONDS),
+ Messages.BucketStoreMessages.UpdateRemoteBuckets.class);
+
+ //Now remove
+ registry1.tell(getRemoveRouteMessage(), mockBroker1.getRef());
+
+ //Bucket store on node2 should get a message to update its local copy of remote buckets
+ probe2.expectMsgClass(
+ FiniteDuration.apply(10, TimeUnit.SECONDS),
+ Messages.BucketStoreMessages.UpdateRemoteBuckets.class);