Java 类akka.actor.TypedProps 实例源码

@BeforeClass
public static void setup() {
    handler.addExceptionElement(e1);
    handler.addExceptionElement(e2);
    s2=new ActorStep("step2",5, null,impl2,handler);
    s1=new ActorStep("step1",5, s2,impl1,handler);
    s1.setWorkRef(0);model.add(s1);
    s2.setWorkRef(1);model.add(s2);
    for(int i=0;i<10;i++){
        data.add(i+1);
    }
    orchestrator =
            TypedActor.get(_system).typedActorOf(
            new TypedProps<OrchestratorImpl>(Orchestrator.class,
            new Creator<OrchestratorImpl>() {
            /**
                 * 
                 */
                private static final long serialVersionUID = 1L;

            public OrchestratorImpl create() { return new OrchestratorImpl(model,batchproducer); }
            }),
            "orchestrator");    
}

public static void main(String[] args) throws Exception {
    ActorSystem _system = ActorSystem.create("TypedActorsExample");

    CalculatorInt calculator1 = TypedActor.get(_system).typedActorOf(
            new TypedProps<Calculator>(CalculatorInt.class,
                    Calculator.class));

    CalculatorInt calculator2 = TypedActor.get(_system).typedActorOf(
            new TypedProps<Calculator>(CalculatorInt.class,
                    Calculator.class));

    // Create a router with Typed Actors
    ActorRef actor1 = TypedActor.get(_system).getActorRefFor(calculator1);
    ActorRef actor2 = TypedActor.get(_system).getActorRefFor(calculator2);

    Iterable<ActorRef> routees = Arrays.asList(new ActorRef[] { actor1,
            actor2 });
    ActorRef router = _system.actorOf(new Props()
            .withRouter(BroadcastRouter.create(routees)));

    router.tell("Hello there");

    _system.shutdown();

}

private <T> T typedActor(final Injector injector, 
    final Class<T> iface, final Class <? extends T> impl) {
  return TypedActor.get(Akka.system()).typedActorOf(
      new TypedProps<T>(iface, new Creator<T>() {
        @Override
        public T create() {
          return injector.getInstance(impl);
        }
      }));
}

public HttpHandler(final ActorSystem system, final ActorRef coordinator) {
    super();
    this.coordinator = coordinator;
    this.system = system;

    requester = TypedActor.get(system).typedActorOf(
            new TypedProps<RequestHandlerActor>(
                    RequestHandlerActor.class,
                    new Creator<RequestHandlerActor>() {
                        public RequestHandlerActor create() {
                            return new RequestHandlerActor(coordinator);
                        }
                    }), "entrypoint");
}

public static void main(String[] args) throws Exception {

        ActorSystem _system = ActorSystem.create("TypedActorsExample",
                ConfigFactory.load().getConfig("TypedActorExample"));

        CalculatorInt calculator = TypedActor.get(_system).typedActorOf(
                new TypedProps<SupervisorActor>(CalculatorInt.class,
                        SupervisorActor.class),"supervisorActor");

        // Get access to the ActorRef
        ActorRef calActor = TypedActor.get(_system).getActorRefFor(calculator);
        // call actor with a message
        calActor.tell("Hi there",calActor);

        //wait for child actor to get restarted
        Thread.sleep(500);

        // Invoke the method and wait for result
        Timeout timeout = new Timeout(Duration.parse("5 seconds"));
        Future<Object> future = Patterns.ask(calActor, Integer.valueOf(10), timeout);
        Integer result = (Integer) Await.result(future, timeout.duration());

        System.out.println("Result from child actor->" + result);

        //wait before shutting down the system
        Thread.sleep(500);

        _system.shutdown();

    }

public static void main(String[] args) throws Exception {
    ActorSystem _system = ActorSystem.create("TypedActorsExample");

    Timeout timeout = new Timeout(Duration.parse("5 seconds"));

    CalculatorInt calculator = TypedActor.get(_system).typedActorOf(
            new TypedProps<Calculator>(CalculatorInt.class,
                    Calculator.class));

    // calling a fire and forget method
    calculator.incrementCount();

    // Invoke the method and wait for result
    Future<Integer> future = calculator.add(Integer.valueOf(14),
            Integer.valueOf(6));
    Integer result = Await.result(future, timeout.duration());

    System.out.println("Result is " + result);

    Option<Integer> counterResult = calculator.incrementAndReturn();
    System.out.println("Result is " + counterResult.get());

    counterResult = calculator.incrementAndReturn();
    System.out.println("Result is " + counterResult.get());

    // Get access to the ActorRef
    ActorRef calActor = TypedActor.get(_system).getActorRefFor(calculator);
    // call actor with a message
    calActor.tell("Hi there");

    _system.shutdown();

}

public static void main(String[] args) throws Exception {
    ActorSystem _system = ActorSystem.create("TypedActorsExample",
            ConfigFactory.load().getConfig("TypedActorExample"));

    Timeout timeout = new Timeout(Duration.parse("5 seconds"));

    CalculatorInt calculator = TypedActor.get(_system).typedActorOf(
            new TypedProps<Calculator>(CalculatorInt.class,
                    Calculator.class).withDispatcher("defaultDispatcher"));

    // calling a fire and forget method
    calculator.incrementCount();

    // Invoke the method and wait for result
    Future<Integer> future = calculator.add(Integer.valueOf(14),
            Integer.valueOf(6));
    Integer result = Await.result(future, timeout.duration());

    System.out.println("Result is " + result);

    Option<Integer> counterResult = calculator.incrementAndReturn();
    System.out.println("Result is " + counterResult.get());

    counterResult = calculator.incrementAndReturn();
    System.out.println("Result is " + counterResult.get());

    // Get access to the ActorRef
    ActorRef calActor = TypedActor.get(_system).getActorRefFor(calculator);
    // call actor with a message
    calActor.tell("Hi there");

    _system.shutdown();

}

@Override
public final <T> T typedActorOf(TypedProps<T> props, String name) {
    return akka.typedActorOf(props, name);
}

@Override
public final <T> T typedActorOf(TypedProps<T> props) {
    return akka.typedActorOf(props);
}

@Override
public final <T> T typedActorOf(TypedProps<T> props, ActorRef actorRef) {
    return akka.typedActorOf(props, actorRef);
}

public TypedActorFactory(Akka akka, TypedProps<T> props) {
    this(akka, props, null);
}

public TypedActorFactory(Akka akka, TypedProps<T> props, String name) {
    this.akka = akka;
    this.props = props;
    this.name = name;
}

public TypedProps<T> create() {
    return TProps.create(interfaceClass, implementationClass);
}

public static TypedProps<Impl> props() {
    return Holder.props;
}

@SuppressWarnings("serial")
public static void main(String[] args) {

    // instanciation of the actor model
    ActorSystem _system = ActorSystem.create("Karajan");
    // Data to be processed
    final List<Integer> data=new ArrayList<Integer>();
    for(int i=0;i<10;i++){
        data.add(i+1);
    }
    // Some implementations
    String impl1="com.wordline.awltech.karajan.orchestrator.masterslavepullpatterntest.Implementation1";
    String impl2="com.wordline.awltech.karajan.orchestrator.masterslavepullpatterntest.Implementation2"; 
    // Somme Error Handling
    ExceptionElement e1=new ExceptionElement("ProcessorException", ErrorStrategy.ONE, Action.SKIP, 5);
    ExceptionElement e2=new ExceptionElement("ArithmeticException", ErrorStrategy.ONE, Action.SKIP, 5);
    ErrorHandling handler=new  ErrorHandling();
    handler.addExceptionElement(e1);
    handler.addExceptionElement(e2);
    // instanciation of some steps
    ActorStep s2=new ActorStep("step2",5, null,impl2,handler);
    ActorStep s1=new ActorStep("step1",5, s2,impl1,handler);
    //Model
    final List<ActorStep> model=new ArrayList<ActorStep>();
    s1.setWorkRef(0);model.add(s1);
    s2.setWorkRef(1);model.add(s2);
      final ActorRef batchproducer =_system.actorOf(Props.create(BatchProducer.class,data.iterator(),5));
    // instanciation of the orchestration
            Orchestrator orchestrator =
                    TypedActor.get(_system).typedActorOf(
                    new TypedProps<OrchestratorImpl>(Orchestrator.class,
                    new Creator<OrchestratorImpl>() {
                    public OrchestratorImpl create() { return new OrchestratorImpl(model,batchproducer); }
                    }),
                    "orchestrator");    
    while(orchestrator.getBatchStatus()!=BatchStatus.COMPLETED){
        System.out.println(orchestrator.getBatchStatus());
    }
    if(orchestrator.getBatchStatus()==BatchStatus.COMPLETED){
        System.out.println("STATUS AT THE END: "+orchestrator.getBatchStatus());
        System.out.println("PROCESSED: "+orchestrator.getStepMetrics("step1").PROCESSED);
        System.out.println("RECEIVED: "+orchestrator.getStepMetrics("step1").RECEIVED);
        _system.shutdown();
    }







}

/**
 * creates a new typed actor
 *
 * @param props the {@link akka.actor.Props} to use to create the actor
 * @param name the name of the actor
 * @param <T> type of the typed actor
 * @return a new typed actor instance
 */
<T> T typedActorOf(TypedProps<T> props, String name);

/**
 * creates a new typed actor that is baked by a normal actor. This is usable if you want to communicate remotely
 * with TypedActors on other machines
 *
 * @param props the {@link akka.actor.Props} to use to create the actor
 * @param actorRef the actor that implements the typed actor
 * @param <T> type of the typed actor
 * @return a new typed actor instance
 */
<T> T typedActorOf(TypedProps<T> props, ActorRef actorRef);

/**
 * creates a new typed actor
 *
 * @param props the {@link akka.actor.Props} to use to create the actor
 * @param <T> type of the typed actor
 * @return a new typed actor instance
 */
<T> T typedActorOf(TypedProps<T> props);

/**
 * creates a new {@link akka.actor.TypedProps} instance.
 *
 * @param interfaceClass the interface class
 * @param implClass the implementation class
 * @param <T> the type of implementation class
 * @return new implementation instance
 */
public static <T> TypedProps<T> create(Class<? super T> interfaceClass, Class<T> implClass) {
    return new TypedProps<T>(interfaceClass, implClass);
}

/**
 * Creates a TypedActor that intercepts the calls and forwards them as {@link akka.actor.TypedActor.MethodCall}
 * to the provided ActorRef.
 */
public <T> T typedActorOf(TypedProps<T> props, String name) {
    return typedActorExtension().typedActorOf(props, name);
}

/**
 * Creates a TypedActor that intercepts the calls and forwards them as {@link akka.actor.TypedActor.MethodCall}
 * to the provided ActorRef.
 */
public <T> T typedActorOf(TypedProps<T> props) {
    return typedActorExtension().typedActorOf(props);
}

/**
 * Creates a TypedActor that intercepts the calls and forwards them as {@link akka.actor.TypedActor.MethodCall}
 * to the provided ActorRef.
 */
public <T> T typedActorOf(TypedProps<T> props, ActorRef actorRef) {
    return typedActorExtension().typedActorOf(props, actorRef);
}