Java 类akka.actor.UntypedActorFactory 实例源码

/** implementation du service Install */   
@SuppressWarnings("serial")
public void install(String nomLogiciel, String listImages) {

    if (nomLogiciel.equals("frascati")) this.logiciel= this.logicielFrascati;
    if (nomLogiciel.equals("nodejs")) this.logiciel= this.logicielNodejs;

    this.imageVirtuels.clear();

    String tab[]= listImages.split(" ");
    for (int i = 0; i < tab.length; i++) {
        this.imageVirtuels.add(new VirtuelImage(tab[i],tab[i],tab[i]));
    }

    // creation de l'actor ControllerSoftwareSupervisor
    controllerSoftwareSupervisor = systemSoftware.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
         return new ControllerSoftwareSupervisor(imageVirtuels,logiciel,sh,new Props(WorkerInstall.class),listener);
        }
    }), "ControllerVMSupervisor");      

    // demarrer le déploiement
    controllerSoftwareSupervisor.tell(new MessageExecuteService()); 

}

/** implementation du service Create container */   
@SuppressWarnings("serial")
public void createContainer(String listImages) {

        System.out.println("nom des container à créer "+listImages);

this.imageVirtuels.clear();
            String tab[]= listImages.split(" ");
            for (int i = 0; i < tab.length; i++) {
                this.imageVirtuels.add(new VirtuelImage(tab[i],tab[i],tab[i])); 
            }

    // creation de l'actor ControllerVMSupervisor
    controllerVmSupervisor = systemVm.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
         return new ControllerVMSupervisor(imageVirtuels,pathFabFile,sh,new Props(WorkerCreateContainer.class),listener);
        }
    }), "ControllerVMSupervisor");      

    // démarrer l'exécution
    controllerVmSupervisor.tell(new MessageExecuteService());   

}

/** implementation du service Create container */   
@SuppressWarnings("serial")
public void startContainer(String listImages) {
this.imageVirtuels.clear();
        String tab[]= listImages.split(" ");
        for (int i = 0; i < tab.length; i++) {
            this.imageVirtuels.add(new VirtuelImage(tab[i],tab[i],tab[i])); 
        }

    // creation de l'actor ControllerVMSupervisor
    controllerVmSupervisor = systemVm.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
         return new ControllerVMSupervisor(imageVirtuels,pathFabFile,sh,new Props(WorkerStartContainer.class),listener);
        }
    }), "ControllerVMSupervisor");      

    // démarrer l'exécution
    controllerVmSupervisor.tell(new MessageExecuteService());   

}

/** implementation du service Create container */   
    @SuppressWarnings("serial")
    public void stopContainer(String listImages) {

this.imageVirtuels.clear();
        String tab[]= listImages.split(" ");
        for (int i = 0; i < tab.length; i++) {
            this.imageVirtuels.add(new VirtuelImage(tab[i],tab[i],tab[i])); 
        }
        // creation de l'actor ControllerVMSupervisor
        controllerVmSupervisor = systemVm.actorOf(new Props(new UntypedActorFactory() {
            public UntypedActor create() {
             return new ControllerVMSupervisor(imageVirtuels,pathFabFile,sh,new Props(WorkerStopContainer.class),listener);
            }
        }), "ControllerVMSupervisor");      

        // démarrer l'exécution
        controllerVmSupervisor.tell(new MessageExecuteService());   

    }

@SuppressWarnings("serial")
public void remoteActorCreationDemo1() {
    log.info("Creating a actor using remote deployment mechanism");

    // create the address object that points to the remote server
    Address addr = new Address("akka", "ServerSys", "127.0.0.1", 2552);

    // creating the ServerActor on the specified remote server
    final ActorRef serverActor = system.actorOf(new Props(ServerActor.class)
            .withDeploy(new Deploy(new RemoteScope(addr))));

    // create a local actor and pass the reference of the remote actor
    actor = system.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
            return new ClientActor(serverActor);
        }
    }));
    // send a message to the local client actor
    actor.tell("Start-RemoteActorCreationDemo1");
}

@SuppressWarnings("serial")
public void remoteActorCreationDemo3() {
    log.info("Creating a actor with remote deployment");

    // creating the ServerActor on the specified remote server
    final ActorRef serverActor = system.actorOf(new Props(ServerActor.class),"remoteServerActor");

    // create a local actor and pass the reference of the remote actor
    actor = system.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
            return new ClientActor(serverActor);
        }
    }));
    // send a message to the local client actor
    actor.tell("Start-RemoteActorCreationDemo3");
}

@Test
public void testFilteringActor() {
    ActorRef filteringActorRef = _system.actorOf(new Props(
            new UntypedActorFactory() {
                public UntypedActor create() {
                    return new FilteringActor(testActor());
                }
            }));
    // pass the reference to implicit sender testActor() otherwise
    // message end up in dead mailbox
    // first test
    filteringActorRef.tell("test message", super.testActor());
    expectMsg("test message");
    // second test
    filteringActorRef.tell(1, super.testActor());
    expectNoMsg();
}

public ApplicationManagerSystem() {

        final int no_of_workers = 10;

        system = ActorSystem.create("LoadGeneratorApp");

        final ActorRef appManager = system.actorOf(
                new Props(new UntypedActorFactory() {
                    public UntypedActor create() {
                        return new JobControllerActor(no_of_msgs);
                    }
                }), "jobController");

        router = system.actorOf(new Props(new UntypedActorFactory() {
            public UntypedActor create() {
                return new WorkerActor(appManager);
            }
        }).withRouter(new RoundRobinRouter(no_of_workers)));
    }

/** implementation du service Pause */   
@SuppressWarnings("serial")
public void pause(String listImages) {


    //Retrait des VM  dans le fichier de configuration du repartiteur (Nginx)
    System.out.println("Connexion sur le serveur de repartition de charge (NGINX) pour retirer les VM dans son fichier de configuration et le signaler");

    this.imageVirtuels.clear();
    String tab[]= listImages.split(" ");
    for (int i = 0; i < tab.length; i++) {
        this.imageVirtuels.add(new VirtuelImage(tab[i],tab[i],tab[i]));     
        //PB gestion des acces concurents sur le fichier a revoir(solution passer plustot la liste des VM)
        String commande="fab -f /home/daniel/scriptFabric/repartiteur/fabfile.py retirerServeur:namehost=172.17.0.30,username=root,passworduser=root,serveur="
            +addressipport(tab[i]);
        sh.system(commande);

    }

    // creation de l'actor ControllerVMSupervisor
    controllerVmSupervisor = systemVm.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
         return new ControllerVMSupervisor(imageVirtuels,pathFabFile,sh,new Props(WorkerPauseVM.class),listener);
        }
    }), "ControllerVMSupervisor");      

    // démarrer l'exécution
    controllerVmSupervisor.tell(new MessageExecuteService());   

}

private void calculate(final int nrOfWorkers, final int nrOfElements, final int nrOfMessage) {
    ActorSystem system = ActorSystem.create("PiSystem");

    final ActorRef listener = system.actorOf(new Props(Listener.class),"listener");

    ActorRef master = system.actorOf(new Props(new UntypedActorFactory() {

        @Override
         public UntypedActor create() {
            return new Master(nrOfWorkers, nrOfMessage, nrOfElements, listener);
          }
    }));

    master.tell(new Calculate());
}

@SuppressWarnings("serial")
public WCMapReduceServer(int no_of_reduce_workers, int no_of_map_workers) {

    system = ActorSystem.create("WCMapReduceApp", ConfigFactory.load()
            .getConfig("WCMapReduceApp"));

    // create the aggregate Actor
    aggregateActor = system.actorOf(new Props(AggregateActor.class));

    // create the list of reduce Actors
    reduceRouter = system.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
            return new ReduceActor(aggregateActor);
        }
    }).withRouter(new RoundRobinRouter(no_of_reduce_workers)));

    // create the list of map Actors
    mapRouter = system.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
            return new MapActor(reduceRouter);
        }
    }).withRouter(new RoundRobinRouter(no_of_map_workers)));

    // create the overall WCMapReduce Actor that acts as the remote actor
    // for clients
    wcMapReduceActor = system.actorOf(new Props(new UntypedActorFactory() {
        public UntypedActor create() {
            return new WCMapReduceActor(aggregateActor, mapRouter);
        }
    }).withDispatcher("priorityMailBox-dispatcher"), "WCMapReduceActor");

}

@Test
public void testForwardingActor() {
    ActorRef forwardingActorRef = _system.actorOf(new Props(
            new UntypedActorFactory() {
                public UntypedActor create() {
                    return new ForwardingActor(testActor());
                }
            }));
    // pass the reference to implicit sender testActor() otherwise
    // message end up in dead mailbox
    forwardingActorRef.tell("test message", super.testActor());
    expectMsg("test message");
}

/**
 * This method it's mirror of the 'main' method, is kept for clarity in parameters usage.
 */
public static void run(WorkflowConfiguration workflowConfiguration, DownloadJobConfiguration configuration) {
  final Injector injector = createInjector(workflowConfiguration, configuration);
  CuratorFramework curator = injector.getInstance(CuratorFramework.class);

  // Create an Akka system
  ActorSystem system = ActorSystem.create("DownloadSystem" + configuration.getDownloadKey());

  // create the master
  ActorRef master = system.actorOf(new Props(new UntypedActorFactory() {
    public UntypedActor create() {
      return injector.getInstance(DownloadMaster.class);
    }
  }), "DownloadMaster" + configuration.getDownloadKey());

  // start the calculation
  master.tell(new DownloadMaster.Start());
  while (!master.isTerminated()) {
    try {
      Thread.sleep(SLEEP_TIME_BEFORE_TERMINATION);
    } catch (InterruptedException ie) {
      LOG.error("Thread interrupted", ie);
    }
  }
  system.shutdown();
  curator.close();
}

/** Run the Monitor. */
    public final void run() {

        systemMonitor = ActorSystem.create("MonitorSystem"); 
        //creation de l'actor Action
        action = systemMonitor.actorOf(new Props(new UntypedActorFactory() {
            public UntypedActor create() {
             return new Action(controllerSoftwareService,controllerVmService,controllerKubectlService,nameReplicationController);
            }
        }), "Actionneur");  
        //creation de l'actor Decide
        decide = systemMonitor.actorOf(new Props(new UntypedActorFactory() {
            public UntypedActor create() {
             return new Decide(action,sh,eventNotificationAlertService);
            }
        }), "Decideur");    
        //creation l'actor Analyse
        analyse = systemMonitor.actorOf(new Props(new UntypedActorFactory() {
            public UntypedActor create() {
             return new Analyse(decide,sh,eventNotificationAlertService);
            }
        }), "Analyseur");

        //creation de l'actor Collect des traces de la version Japonais de Wikipedia
//      collect = systemMonitor.actorOf(new Props(new UntypedActorFactory() {
//          public UntypedActor create() {
//           return new CollectTraceWikipedia(analyse,sh);
//          }
//      }), "Collect");

        //creation de l'actor Collect en temps réel
        collect = systemMonitor.actorOf(new Props(new UntypedActorFactory() {
            public UntypedActor create() {
             return new CollectRealTime(analyse,sh);
            }
        }), "Collect");


          System.out.println("START");
          //lecture du nombre de requettes tous les 30 secondes
          Cancellable cancellable = systemMonitor.scheduler().schedule(Duration.Zero(),Duration.create(30, "seconds"),collect, new Interval(30));                 
          try {
              Thread.sleep(30000000);
            } catch (InterruptedException e) {
                System.out.println(e.getStackTrace());
                return;
            }

          cancellable.cancel();
          System.out.println("END");    
    }

/**
 * Here dont need the nodeGroupMetaData.Assuming the request content has
 * already been ready as in the keys of the hashmap with all the target
 * nodes. So just need a string to have the nodeGroupType is fine.
 * 
 * @param nodeGroupType
 * @param agentCommandType
 * @param dataStore
 * @return
 */
public void sendAggregationCommandToManager(String patternStr,
        AggregateData aggregateData) {

    ActorRef aggregationManager = null;
    try {
        // Start new job
        String directorJobUuid = UUID.randomUUID().toString();

        models.utils.LogUtils.printLogNormal("!!STARTED sendAggregationCommandToManager : "
                + directorJobUuid + " at " + DateUtils.getNowDateTimeStr());

        // Get the singleton actor system

        // create the master
        aggregationManager = ActorConfig.getActorSystem().actorOf(
                new Props(new UntypedActorFactory() {
                    private static final long serialVersionUID = 1L;

                    public UntypedActor create() {
                        return new AggregationManager();
                    }
                }), "AggregationManager-" + UUID.randomUUID().toString());

        final FiniteDuration duration = Duration.create(
                VarUtils.TIMEOUT_ASK_AGGREGATION_MANAGER_SCONDS,
                TimeUnit.SECONDS);
        Future<Object> future = Patterns.ask(aggregationManager,
                new RequestToAggregationManager(patternStr,
                        directorJobUuid, aggregateData), new Timeout(
                        duration));

        responseFromAggregationManager = (ResponseFromAggregationManager) Await
                .result(future, duration);

        models.utils.LogUtils.printLogNormal("!!COMPLETED sendAggregationCommandToManager : "
                + directorJobUuid + " at " + DateUtils.getNowDateTimeStr());

    } catch (Throwable ex) {
         models.utils.LogUtils.printLogError
                 ("Exception in sendAggregationCommandToManager : "
                        + ex.getLocalizedMessage());
        ex.printStackTrace();
    } finally {

        // stop the manager:agentCommandManager
        if (aggregationManager != null) {
            ActorConfig.getActorSystem().stop(aggregationManager);
        }

        aggregationManager = null;

    }
}

private final void processRequest() {

        // the first time dont count. the first time will delay
        if (!hasBeenDelayed) {

            // jeff only the first time will pause

            sender = getSender();
            startTimeMillis = System.currentTimeMillis();

            validateRequest(request);
            timeoutDuration = Duration.create(
                    request.getMaxOperationTimeSeconds(), TimeUnit.SECONDS);

            hasBeenDelayed = true;
            /**
             * 20131013 if it is 0; no need to schedule another message.
             */
            if (this.request.getPauseIntervalBeforeSendMillis() != 0L) {
                long MAX_PAUSE_INTERVAL_MILLIS = 600000L; // 600 sec
                long pauseIntervalWorkerMillis = Math.min(
                        MAX_PAUSE_INTERVAL_MILLIS,
                        this.request.getPauseIntervalBeforeSendMillis());
                getContext()
                        .system()
                        .scheduler()
                        .scheduleOnce(
                                Duration.create(pauseIntervalWorkerMillis,
                                        TimeUnit.MILLISECONDS),
                                getSelf(),
                                OperationWorker.MessageType.PROCESS_REQUEST,
                                getContext().system().dispatcher());

                return;

            }
        }

        /**
         * 20130917: change to add uniform target node capability
         */

        final String trueTargetNode = (hostUniform == null) ? host
                : hostUniform;

        asyncWorker = getContext().actorOf(new Props(new UntypedActorFactory() {
            private static final long serialVersionUID = 1L;

            public Actor create() {
                final String requestUrl = String.format("%s://%s:%d%s",
                        protocol.toString(), trueTargetNode, agentPort,
                        request.getResourcePath());

                return new HttpWorker(client, protocol, requestUrl,
                        request.getHttpMethod(), request.getPostData(),
                        CONTENT_TYPE_JSON, HTTP_MAX_RETRIES,
                        HTTP_RETRY_INTERVAL_MILLIS, httpHeaderType);
            }
        }));
        asyncWorker.tell(HttpWorker.MessageType.PROCESS_REQUEST,
                getSelf());

        // To handle cases where this operation takes extremely long, schedule a
        // 'timeout' message to be sent to us
        timeoutMessageCancellable = getContext()
                .system()
                .scheduler()
                .scheduleOnce(timeoutDuration, getSelf(),
                        InternalMessageType.OPERATION_TIMEOUT,
                        getContext().system().dispatcher());

    }

public SpringUntypedActorFactory(UntypedActorFactory customFactory, ApplicationContext applicationContext) {
    this.dependencyInjectionFactory = new SpecificUntypedActorFactory(customFactory);
    this.applicationContext = applicationContext;
}

private SpecificUntypedActorFactory(UntypedActorFactory specificFactory) {
    this.specificFactory = specificFactory;
}

public DependencyInjectionProps(ApplicationContext applicationContext, UntypedActorFactory factory) {
    super(new SpringUntypedActorFactory(factory, applicationContext));
}

@SuppressWarnings("serial")
public WorkServerActorSystem() {
    // load the configuration
    system = ActorSystem.create("WorkServerSys", ConfigFactory.load()
            .getConfig("WorkServerSys"));
    log = Logging.getLogger(system, this);

    // create the work scheduler actor
    workSchedulerActor = system.actorOf(
            new Props(WorkSchedulerActor.class), "WorkSchedulerActor");

    // create the job controller actor, which manages the routees and sends
    // out
    // work packets to the registered workers
    jobControllerActor = system.actorOf(new Props(
            new UntypedActorFactory() {
                public UntypedActor create() {
                    return new JobControllerActor(workSchedulerActor);
                }
            }), "JobControllerActor");

    remoteActorListener = system.actorOf(new Props(
            new UntypedActorFactory() {
                public UntypedActor create() {
                    return new RemoteClientEventListener(jobControllerActor);
                }
            }), "RemoteClientEventListener");


    // actor that registers and unregisters the workers
    registerRemoteWorkerActor = system.actorOf(new Props(
            new UntypedActorFactory() {
                public UntypedActor create() {
                    return new RegisterRemoteWorkerActor(jobControllerActor);
                }
            }), "RegisterRemoteWorkerActor");

    workSchedulerActor.tell("Start Sending Work", jobControllerActor);

    system.eventStream().subscribe(remoteActorListener,
            RemoteLifeCycleEvent.class);

}