+ private void receiveGetRouter(final FindRouters findRouters) {
+ log.debug("receiveGetRouter for {}", findRouters.getRouteIdentifier());
+
+ final ActorRef sender = getSender();
+ if (!findRouters(findRouters, sender)) {
+ log.debug("No routers found for {} - scheduling {} ms timer", findRouters.getRouteIdentifier(),
+ findRouterTimeout.toMillis());
+
+ final AtomicReference<Cancellable> timer = new AtomicReference<>();
+ final Runnable routesUpdatedRunnable = new Runnable() {
+ @Override
+ public void run() {
+ if (findRouters(findRouters, sender)) {
+ routesUpdatedCallbacks.remove(this);
+ timer.get().cancel();
+ }
+ }
+ };
+
+ routesUpdatedCallbacks.add(routesUpdatedRunnable);
+
+ Runnable timerRunnable = () -> {
+ log.warn("Timed out finding routers for {}", findRouters.getRouteIdentifier());
+
+ routesUpdatedCallbacks.remove(routesUpdatedRunnable);
+ sender.tell(new Messages.FindRoutersReply(
+ Collections.<Pair<ActorRef, Long>>emptyList()), self());
+ };
+
+ timer.set(getContext().system().scheduler().scheduleOnce(findRouterTimeout, self(), timerRunnable,
+ getContext().dispatcher(), self()));
+ }
+ }
+
+ private boolean findRouters(FindRouters findRouters, ActorRef sender) {