raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Applied discussion results to implementation and documentation.

Browse source 
- renamed to SupervisedAsk.askOf
- watches for actor termination
- askOf returns Future
- intermediate actor used to prevent actor system bottle-neck
- timeout handling
This commit is contained in:
RickLatrine 2012-10-13 16:24:27 +02:00
parent 1b34290d4c
commit b965506f10
3 changed files with 101 additions and 104 deletions
Download patch file Download diff file Expand all files Collapse all files

45
akka-docs/rst/java/code/docs/pattern/JavaAsk.java
View file

@ -1,45 +0,0 @@
public class JavaAsk extends UntypedActor {
private ActorRef targetActor;
private ActorRef caller;
private static class AskParam {
Props props;
Object message;
AskParam(Props props, Object message) {
this.props = props;
this.message = message;
}
}
@Override
public SupervisorStrategy supervisorStrategy() {
return new OneForOneStrategy(0, Duration.Zero(), new Function<Throwable, Directive>() {
public Directive apply(Throwable cause) {
caller.tell(new Status.Failure(cause));
return SupervisorStrategy.stop();
}
});
}
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof AskParam) {
AskParam param = (AskParam) message;
caller = getSender();
targetActor = getContext().actorOf(param.props);
targetActor.forward(param.message, getContext());
} else
unhandled(message);
}
public static void ask(ActorSystem system, Props props, Object message, Timeout timeout) throws Exception {
ActorRef javaAsk = system.actorOf(Props.apply(JavaAsk.class));
try {
AskParam param = new AskParam(props, message);
Future<Object> finished = Patterns.ask(javaAsk, param, timeout);
Await.result(finished, timeout.duration());
} finally {
system.stop(javaAsk);
}
}
}

90
akka-docs/rst/java/code/docs/pattern/SupervisedAsk.java Normal file
View file

@ -0,0 +1,90 @@
public class SupervisedAsk {
private static class AskParam {
Props props;
Object message;
Timeout timeout;
AskParam(Props props, Object message, Timeout timeout) {
this.props = props;
this.message = message;
this.timeout = timeout;
}
}
private static class AskTimeout {}
public static class AskSupervisorCreator extends UntypedActor {
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof AskParam) {
ActorRef supervisor = getContext().actorOf(
Props.apply(AskSupervisor.class));
supervisor.forward(message, getContext());
} else {
unhandled(message);
}
}
}
public static class AskSupervisor extends UntypedActor {
private ActorRef targetActor;
private ActorRef caller;
private AskParam askParam;
private Cancellable timeoutMessage;
@Override
public SupervisorStrategy supervisorStrategy() {
return new OneForOneStrategy(0, Duration.Zero(),
new Function<Throwable, Directive>() {
public Directive apply(Throwable cause) {
caller.tell(new Status.Failure(cause));
return SupervisorStrategy.stop();
}
});
}
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof AskParam) {
askParam = (AskParam) message;
caller = getSender();
targetActor = getContext().actorOf(askParam.props);
getContext().watch(targetActor);
targetActor.forward(askParam.message, getContext());
Scheduler scheduler = getContext().system().scheduler();
timeoutMessage = scheduler.scheduleOnce(askParam.timeout.duration(), self(), new AskTimeout());
} else if (message instanceof Terminated) {
Throwable ex = new ActorKilledException("Target actor terminated.");
caller.tell(new Status.Failure(ex));
timeoutMessage.cancel();
getContext().stop(self());
} else if (message instanceof AskTimeout) {
Throwable ex = new TimeoutException("Target actor timed out after " + askParam.timeout.toString());
caller.tell(new Status.Failure(ex));
getContext().stop(self());
} else
unhandled(message);
}
}
public static Future<Object> askOf(ActorRef supervisorCreator,
Props props, Object message, Timeout timeout) {
AskParam param = new AskParam(props, message, timeout);
return Patterns.ask(supervisorCreator, param, timeout);
}
synchronized public static ActorRef createSupervisorCreator(ActorRefFactory factory) {
return factory.actorOf(Props.apply(AskSupervisorCreator.class));
}
}
// example usage
try {
ActorRef supervisorCreator = SupervisedAsk.createSupervisorCreator(actorSystem);
Future<Object> finished = SupervisedAsk.askOf(supervisorCreator,
Props.apply(SomeActor.class), message, timeout);
SomeResult result = (SomeResult) Await.result(finished, timeout.duration());
} catch (Exception e) {
// exception propagated by supervision
}

70
akka-docs/rst/java/howto.rst
View file

@ -16,8 +16,8 @@ sense to add to the ``akka.pattern`` package for creating an `OTP-like library
You might find some of the patterns described in the Scala chapter of
:ref:`howto-scala` useful even though the example code is written in Scala.
Exception propagation in actor hierarchy using supervision
==========================================================
Single-Use Actor Trees with High-Level Error Reporting
======================================================
*Contributed by: Rick Latrine*
@ -26,7 +26,7 @@ This method starts a temporary actor to forward the message and collect the resu
In case of errors within the asked actor the default supervision handling will take over.
The caller of Patterns.ask() will not be notified.
If that caller is interested in such an exception, he must reply to the temporary actor with Status.Failure(Throwable).
If that caller is interested in such an exception, he must make sure that the asked actor replies with Status.Failure(Throwable).
Behind the asked actor a complex actor hierarchy might be spawned to accomplish asynchronous work.
Then supervision is the established way to control error handling.
@ -34,70 +34,22 @@ Unfortunately the asked actor must know about supervision and must catch the exc
Such an actor is unlikely to be reused in a different actor hierarchy and contains crippled try/catch blocks.
This pattern provides a way to encapsulate supervision and error propagation to the temporary actor.
Finally the exception occurred in the asked actor is thrown by Patterns.ask().
Finally the promise returned by Patterns.ask() is fulfilled as a failure, including the exception.
Let's have a look at the example code:
.. code-block:: java
.. includecode:: code/docs/pattern/SupervisedAsk.java
public class JavaAsk extends UntypedActor {
private ActorRef targetActor;
private ActorRef caller;
private static class AskParam {
Props props;
Object message;
AskParam(Props props, Object message) {
this.props = props;
this.message = message;
}
}
@Override
public SupervisorStrategy supervisorStrategy() {
return new OneForOneStrategy(0, Duration.Zero(), new Function<Throwable, Directive>() {
public Directive apply(Throwable cause) {
caller.tell(new Status.Failure(cause));
return SupervisorStrategy.stop();
}
});
}
@Override
public void onReceive(Object message) throws Exception {
if (message instanceof AskParam) {
AskParam param = (AskParam) message;
caller = getSender();
targetActor = getContext().actorOf(param.props);
targetActor.forward(param.message, getContext());
} else
unhandled(message);
}
public static void ask(ActorSystem system, Props props, Object message, Timeout timeout) throws Exception {
ActorRef javaAsk = system.actorOf(Props.apply(JavaAsk.class));
try {
AskParam param = new AskParam(props, message);
Future<Object> finished = Patterns.ask(javaAsk, param, timeout);
Await.result(finished, timeout.duration());
} finally {
system.stop(javaAsk);
}
}
}
}
In the ask method the parent actor is started.
The parent is sent which child it should create and which message is to be forwarded.
On receiving these parameteres the parent actor creates the target/child actor.
The message is forwarded in order to let the child actor reply the result directly to the temporary actor.
In the askOf method the SupervisorCreator is sent the user message.
The SupervisorCreator creates a SupervisorActor and forwards the message.
This prevents the actor system from overloading due to actor creations.
The SupervisorActor is responsible to create the user actor, forwards the message, handles actor termination and supervision.
Additionally the SupervisorActor stops the user actor if execution time expired.
In case of an exception the supervisor tells the temporary actor which exception was thrown.
Afterwards the actor hierarchy is stopped.
The exception will be thrown by the ask method.
Finally we are able to execute an actor and receive the results or exceptions synchronously.
Finally we are able to execute an actor and receive the results or exceptions.
Template Pattern