import akka.routing.ActorRefRoutee; //导入依赖的package包/类
/** partitions the market by product ID and creates an actor encapsulating an engine, per 
 * partition. returns a router containing all the kids together with the suitable logic 
 * to route to the correct engine. */
private Router partitionAndCreateRouter() {
	Map<String, ActorRef> kids = new HashMap<>();
	java.util.List<Routee> routees = new ArrayList<Routee>();
	int chunk = Constants.PRODUCT_IDS.length / NUM_KIDS;
	for (int i = 0, j = Constants.PRODUCT_IDS.length; i < j; i += chunk) {
	    String[] temparray = Arrays.copyOfRange(Constants.PRODUCT_IDS, i, i + chunk);
	    LOGGER.info("created engine for products " + temparray);
	    ActorRef actor = getContext().actorOf(Props.create(EngineActor.class));
	    getContext().watch(actor);
	    routees.add(new ActorRefRoutee(actor));

	    for (int k = 0; k < temparray.length; k++) {
	    	LOGGER.debug("mapping productId '" + temparray[k] + "' to engine " + i);
	    	kids.put(temparray[k], actor);
	    }
	    LOGGER.info("---started trading");
	    actor.tell(EngineActor.RUN, ActorRef.noSender());
	}			
	Router router = new Router(new PartitioningRoutingLogic(kids), routees);
	return router;
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public Routee select(Object message, IndexedSeq<Routee> routees) {

	//find which product ID is relevant here
	String productId = null;
	if(message instanceof PurchaseOrder){
		productId = ((PurchaseOrder) message).getProductId();
	}else if(message instanceof SalesOrder){
		productId = ((SalesOrder) message).getProductId();
	}
	ActorRef actorHandlingProduct = kids.get(productId);

	//no go find the routee for the relevant actor
	for(Routee r : JavaConversions.asJavaIterable(routees)){
		ActorRef a = ((ActorRefRoutee) r).ref(); //cast ok, since the are by definition in this program all routees to ActorRefs
		if(a.equals(actorHandlingProduct)){
			return r;
		}
	}
	
	return akka.routing.NoRoutee$.MODULE$; //none found, return NoRoutee
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public void preStart() {
    Optional<PeerURLType> peerURL = discoveryConfiguration.getDiscoveryURL()
            .stream()
            .filter((url) -> url.getType().equalsIgnoreCase(NsiConstants.NSI_TOPOLOGY_V1))
            .findFirst();

    isConfigured = peerURL.isPresent();
    if (isConfigured) {
        manifestReader.setTarget(peerURL.get().getValue());
    }
    else {
        log.info("AgoleDiscoveryRouter: No AGOLE URL provisioned so disabling audit.");
        return;
    }

    List<Routee> routees = new ArrayList<>();
    for (int i = 0; i < getPoolSize(); i++) {
        ActorRef r = getContext().actorOf(Props.create(AgoleDiscoveryActor.class));
        getContext().watch(r);
        routees.add(new ActorRefRoutee(r));
    }
    router = new Router(new RoundRobinRoutingLogic(), routees);
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public void preStart() {
    List<Routee> routees = new ArrayList<>();
    for (int i = 0; i < getPoolSize(); i++) {
        ActorRef r = getContext().actorOf(Props.create(RegistrationActor.class, discoveryConfiguration, remoteSubscriptionCache));
        getContext().watch(r);
        routees.add(new ActorRefRoutee(r));
    }
    router = new Router(new RoundRobinRoutingLogic(), routees);
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public void preStart() {
    List<Routee> routees = new ArrayList<>();
    for (int i = 0; i < getPoolSize(); i++) {
        ActorRef r = getContext().actorOf(Props.create(NotificationActor.class, discoveryConfiguration.getNsaId(), restClient));
        getContext().watch(r);
        routees.add(new ActorRefRoutee(r));
    }
    router = new Router(new RoundRobinRoutingLogic(), routees);
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public void onReceive(Object msg) {
    if (msg instanceof DocumentEvent) {
        // We have a document event.
        DocumentEvent de = (DocumentEvent) msg;
        log.debug("NotificationRouter: document event {}, id={}", de.getEvent(), de.getDocument().getId());
        routeDocumentEvent(de);
    }
    else if (msg instanceof SubscriptionEvent) {
        // We have a subscription event.
        SubscriptionEvent se = (SubscriptionEvent) msg;
        log.debug("NotificationRouter: subscription event id={}, requesterId={}", se.getSubscription().getId(), se.getSubscription().getSubscription().getRequesterId());
        routeSubscriptionEvent(se);
    }
    else if (msg instanceof Terminated) {
        log.debug("NotificationRouter: terminate event.");
        router = router.removeRoutee(((Terminated) msg).actor());
        ActorRef r = getContext().actorOf(Props.create(NotificationActor.class));
        getContext().watch(r);
        router = router.addRoutee(new ActorRefRoutee(r));
    }
    else if (msg instanceof StartMsg) {
        // We ignore these for now as we have no specific start task.
    }
    else {
        log.debug("NotificationRouter: unhandled event = {}", msg.getClass().getName());
        unhandled(msg);
    }
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public void preStart() {
    List<Routee> routees = new ArrayList<>();
    for (int i = 0; i < getPoolSize(); i++) {
        ActorRef r = getContext().actorOf(Props.create(Gof3DiscoveryActor.class));
        getContext().watch(r);
        routees.add(new ActorRefRoutee(r));
    }
    router = new Router(new RoundRobinRoutingLogic(), routees);
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public void onReceive(Object msg) {
    TimerMsg message = new TimerMsg();

    // Check to see if we got the go ahead to start registering.
    if (msg instanceof StartMsg) {
        // Create a Register event to start us off.
        msg = message;
    }

    if (msg instanceof TimerMsg) {
        routeTimerEvent();
        ddsActorSystem.getActorSystem().scheduler().scheduleOnce(Duration.create(getInterval(), TimeUnit.SECONDS), this.getSelf(), message, ddsActorSystem.getActorSystem().dispatcher(), null);
    }
    else if (msg instanceof Gof3DiscoveryMsg) {
        Gof3DiscoveryMsg incoming = (Gof3DiscoveryMsg) msg;

        log.debug("onReceive: discovery update for nsaId=" + incoming.getNsaId());

        discovery.put(incoming.getNsaURL(), incoming);
    }
    else if (msg instanceof Terminated) {
        log.debug("onReceive: terminate event.");
        router = router.removeRoutee(((Terminated) msg).actor());
        ActorRef r = getContext().actorOf(Props.create(Gof3DiscoveryActor.class));
        getContext().watch(r);
        router = router.addRoutee(new ActorRefRoutee(r));
    }
    else {
        log.error("onReceive: unhandled event.");
        unhandled(msg);
    }
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
private List<Routee> createParserActorBroadcastRoutes(List<ResponseParser> responseParsers) {
    return responseParsers.stream().map(responseParser -> {
        final ActorRef parserActorRef = getContext().actorOf(ParserActor.props(responseParser));
        return new ActorRefRoutee(parserActorRef);
    }).collect(Collectors.toList());
}
 

import akka.routing.ActorRefRoutee; //导入依赖的package包/类
@Override
public void onReceive(Object msg) {
    TimerMsg message = new TimerMsg();

    // Check to see if we got the go ahead to start registering.
    if (msg instanceof StartMsg) {
        // Create a Register event to start us off.
        msg = message;
        if (!isConfigured) {
            log.info("onReceive: StartMsg no AGOLE URL provisioned so disabling audit.");
            return;
        }
    }

    if (msg instanceof TimerMsg) {
        log.debug("onReceive: timer event.");
        if (!isConfigured) {
            log.info("onReceive: TimerMsg no AGOLE URL provisioned so disabling audit.");
            return;
        }
        if (readManifest() != null) {
            routeTimerEvent();
        }

        ddsActorSystem.getActorSystem().scheduler().scheduleOnce(Duration.create(getInterval(), TimeUnit.SECONDS), this.getSelf(), message, ddsActorSystem.getActorSystem().dispatcher(), null);
    }
    else if (msg instanceof AgoleDiscoveryMsg) {
        AgoleDiscoveryMsg incoming = (AgoleDiscoveryMsg) msg;

        log.debug("onReceive: discovery update for nsaId={}", incoming.getNsaId());

        discovery.put(incoming.getTopologyURL(), incoming);
    }
    else if (msg instanceof Terminated) {
        log.debug("onReceive: terminate event.");
        if (router != null) {
            router = router.removeRoutee(((Terminated) msg).actor());
            ActorRef r = getContext().actorOf(Props.create(AgoleDiscoveryActor.class));
            getContext().watch(r);
            router = router.addRoutee(new ActorRefRoutee(r));
        }
    }
    else {
        log.debug("onReceive: unhandled event.");
        unhandled(msg);
    }
}