Merge branch 'wip-1456-document-typed-actors-√'
This commit is contained in:
Viktor Klang
commit
6c96397711
16 changed files with 632 additions and 418 deletions
|
|
@ -99,7 +99,7 @@ object TypedActorSpec {
|
|||
}
|
||||
|
||||
def futureComposePigdogFrom(foo: Foo): Future[String] = {
|
||||
implicit val timeout = TypedActor.system.settings.ActorTimeout
|
||||
implicit val timeout = TypedActor.context.system.settings.ActorTimeout
|
||||
foo.futurePigdog(500).map(_.toUpperCase)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -51,6 +51,8 @@ trait TypedActorFactory {
|
|||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the suppli ed interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*
|
||||
* Java API
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Class[T], props: Props): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.newInstance, props, None, interface.getClassLoader)
|
||||
|
|
@ -59,6 +61,8 @@ trait TypedActorFactory {
|
|||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*
|
||||
* Java API
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Class[T], props: Props, name: String): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.newInstance, props, Some(name), interface.getClassLoader)
|
||||
|
|
@ -67,6 +71,8 @@ trait TypedActorFactory {
|
|||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*
|
||||
* Java API
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Creator[T], props: Props): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.create, props, None, interface.getClassLoader)
|
||||
|
|
@ -75,6 +81,8 @@ trait TypedActorFactory {
|
|||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*
|
||||
* Java API
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Creator[T], props: Props, name: String): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.create, props, Some(name), interface.getClassLoader)
|
||||
|
|
@ -83,55 +91,21 @@ trait TypedActorFactory {
|
|||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*
|
||||
* Scala API
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Class[T], props: Props, loader: ClassLoader): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.newInstance, props, None, loader)
|
||||
|
||||
/**
|
||||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Class[T], props: Props, name: String, loader: ClassLoader): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.newInstance, props, Some(name), loader)
|
||||
|
||||
/**
|
||||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Creator[T], props: Props, loader: ClassLoader): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.create, props, None, loader)
|
||||
|
||||
/**
|
||||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied interface class (if the class represents an interface) or
|
||||
* all interfaces (Class.getInterfaces) if it's not an interface class
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: Creator[T], props: Props, name: String, loader: ClassLoader): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl.create, props, Some(name), loader)
|
||||
def typedActorOf[R <: AnyRef, T <: R](interface: Class[R], impl: ⇒ T, props: Props, name: String): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, interface, impl, props, Some(name), interface.getClassLoader)
|
||||
|
||||
/**
|
||||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied implementation class' interfaces (Class.getInterfaces)
|
||||
*
|
||||
* Scala API
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](impl: Class[T], props: Props, loader: ClassLoader): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, impl, impl.newInstance, props, None, loader)
|
||||
|
||||
/**
|
||||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied implementation class' interfaces (Class.getInterfaces)
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](impl: Class[T], props: Props, name: String, loader: ClassLoader): R =
|
||||
typedActor.createProxyAndTypedActor(actorFactory, impl, impl.newInstance, props, Some(name), loader)
|
||||
|
||||
/**
|
||||
* Creates a new TypedActor proxy using the supplied Props,
|
||||
* the interfaces usable by the returned proxy is the supplied implementation class' interfaces (Class.getInterfaces)
|
||||
*/
|
||||
def typedActorOf[R <: AnyRef, T <: R](props: Props = Props(), name: String = null, loader: ClassLoader = null)(implicit m: Manifest[T]): R = {
|
||||
val clazz = m.erasure.asInstanceOf[Class[T]]
|
||||
typedActor.createProxyAndTypedActor(actorFactory, clazz, clazz.newInstance, props, Option(name), if (loader eq null) clazz.getClassLoader else loader)
|
||||
def typedActorOf[R <: AnyRef, T <: R: ClassManifest](props: Props = Props(), name: String = null): R = {
|
||||
val clazz = implicitly[ClassManifest[T]].erasure.asInstanceOf[Class[T]]
|
||||
typedActor.createProxyAndTypedActor(actorFactory, clazz, clazz.newInstance, props, Option(name), clazz.getClassLoader)
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -172,6 +146,8 @@ trait TypedActorFactory {
|
|||
}
|
||||
|
||||
object TypedActor extends ExtensionId[TypedActorExtension] with ExtensionIdProvider {
|
||||
override def get(system: ActorSystem): TypedActorExtension = super.get(system)
|
||||
|
||||
def lookup() = this
|
||||
def createExtension(system: ActorSystemImpl): TypedActorExtension = new TypedActorExtension(system)
|
||||
|
||||
|
|
@ -255,7 +231,7 @@ object TypedActor extends ExtensionId[TypedActorExtension] with ExtensionIdProvi
|
|||
}
|
||||
|
||||
private val selfReference = new ThreadLocal[AnyRef]
|
||||
private val currentSystem = new ThreadLocal[ActorSystem]
|
||||
private val currentContext = new ThreadLocal[ActorContext]
|
||||
|
||||
/**
|
||||
* Returns the reference to the proxy when called inside a method call in a TypedActor
|
||||
|
|
@ -281,23 +257,30 @@ object TypedActor extends ExtensionId[TypedActorExtension] with ExtensionIdProvi
|
|||
}
|
||||
|
||||
/**
|
||||
* Returns the akka system (for a TypedActor) when inside a method call in a TypedActor.
|
||||
* Returns the ActorContext (for a TypedActor) when inside a method call in a TypedActor.
|
||||
*/
|
||||
def system = currentSystem.get match {
|
||||
case null ⇒ throw new IllegalStateException("Calling TypedActor.system outside of a TypedActor implementation method!")
|
||||
def context = currentContext.get match {
|
||||
case null ⇒ throw new IllegalStateException("Calling TypedActor.context outside of a TypedActor implementation method!")
|
||||
case some ⇒ some
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the default dispatcher (for a TypedActor) when inside a method call in a TypedActor.
|
||||
*/
|
||||
implicit def dispatcher = system.dispatcher
|
||||
implicit def dispatcher = context.dispatcher
|
||||
|
||||
/**
|
||||
* Implementation of TypedActor as an Actor
|
||||
*/
|
||||
private[akka] class TypedActor[R <: AnyRef, T <: R](val proxyVar: AtomVar[R], createInstance: ⇒ T) extends Actor {
|
||||
val me = createInstance
|
||||
val me = try {
|
||||
TypedActor.selfReference set proxyVar.get
|
||||
TypedActor.currentContext set context
|
||||
createInstance
|
||||
} finally {
|
||||
TypedActor.selfReference set null
|
||||
TypedActor.currentContext set null
|
||||
}
|
||||
|
||||
override def preStart(): Unit = me match {
|
||||
case l: PreStart ⇒ l.preStart()
|
||||
|
|
@ -331,7 +314,7 @@ object TypedActor extends ExtensionId[TypedActorExtension] with ExtensionIdProvi
|
|||
def receive = {
|
||||
case m: MethodCall ⇒
|
||||
TypedActor.selfReference set proxyVar.get
|
||||
TypedActor.currentSystem set context.system
|
||||
TypedActor.currentContext set context
|
||||
try {
|
||||
if (m.isOneWay) m(me)
|
||||
else {
|
||||
|
|
@ -351,7 +334,7 @@ object TypedActor extends ExtensionId[TypedActorExtension] with ExtensionIdProvi
|
|||
}
|
||||
} finally {
|
||||
TypedActor.selfReference set null
|
||||
TypedActor.currentSystem set null
|
||||
TypedActor.currentContext set null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -45,6 +45,9 @@ object Await {
|
|||
def result[T](awaitable: Awaitable[T], atMost: Duration): T = awaitable.result(atMost)
|
||||
}
|
||||
|
||||
/**
|
||||
* Futures is the Java API for Futures and Promises
|
||||
*/
|
||||
object Futures {
|
||||
|
||||
/**
|
||||
|
|
@ -57,6 +60,16 @@ object Futures {
|
|||
*/
|
||||
def promise[T](dispatcher: MessageDispatcher): Promise[T] = Promise[T]()(dispatcher)
|
||||
|
||||
/**
|
||||
* Java API, creates an already completed Promise with the specified exception
|
||||
*/
|
||||
def failed[T](exception: Throwable, dispatcher: MessageDispatcher): Promise[T] = Promise.failed(exception)(dispatcher)
|
||||
|
||||
/**
|
||||
* Java API, Creates an already completed Promise with the specified result
|
||||
*/
|
||||
def successful[T](result: T, dispatcher: MessageDispatcher): Promise[T] = Promise.successful(result)(dispatcher)
|
||||
|
||||
/**
|
||||
* Java API.
|
||||
* Returns a Future that will hold the optional result of the first Future with a result that matches the predicate
|
||||
|
|
|
|||
|
|
@ -3,9 +3,10 @@
|
|||
*/
|
||||
package akka.serialization
|
||||
|
||||
import akka.actor.{ ExtensionId, ExtensionIdProvider, ActorSystemImpl }
|
||||
import akka.actor.{ ActorSystem, ExtensionId, ExtensionIdProvider, ActorSystemImpl }
|
||||
|
||||
object SerializationExtension extends ExtensionId[Serialization] with ExtensionIdProvider {
|
||||
override def get(system: ActorSystem): Serialization = super.get(system)
|
||||
override def lookup = SerializationExtension
|
||||
override def createExtension(system: ActorSystemImpl): Serialization = new Serialization(system)
|
||||
}
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
package akka.docs.actor
|
||||
|
||||
import org.scalatest.junit.JUnitSuite
|
||||
|
||||
class TypedActorDocTest extends TypedActorDocTestBase with JUnitSuite
|
||||
150
akka-docs/java/code/akka/docs/actor/TypedActorDocTestBase.java
Normal file
150
akka-docs/java/code/akka/docs/actor/TypedActorDocTestBase.java
Normal file
|
|
@ -0,0 +1,150 @@
|
|||
package akka.docs.actor;
|
||||
|
||||
//#imports
|
||||
import akka.dispatch.*;
|
||||
import akka.actor.*;
|
||||
import akka.japi.*;
|
||||
import akka.util.Duration;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
//#imports
|
||||
import java.lang.Exception;
|
||||
|
||||
import org.junit.Test;
|
||||
import static org.junit.Assert.*;
|
||||
public class TypedActorDocTestBase {
|
||||
Object someReference = null;
|
||||
ActorSystem system = null;
|
||||
|
||||
//#typed-actor-iface
|
||||
public static interface Squarer {
|
||||
//#typed-actor-iface-methods
|
||||
void squareDontCare(int i); //fire-forget
|
||||
|
||||
Future<Integer> square(int i); //non-blocking send-request-reply
|
||||
|
||||
Option<Integer> squareNowPlease(int i);//blocking send-request-reply
|
||||
|
||||
int squareNow(int i); //blocking send-request-reply
|
||||
//#typed-actor-iface-methods
|
||||
}
|
||||
//#typed-actor-iface
|
||||
|
||||
//#typed-actor-impl
|
||||
static class SquarerImpl implements Squarer {
|
||||
private String name;
|
||||
|
||||
public SquarerImpl() {
|
||||
this.name = "default";
|
||||
}
|
||||
|
||||
public SquarerImpl(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
//#typed-actor-impl-methods
|
||||
|
||||
public void squareDontCare(int i) {
|
||||
int sq = i * i; //Nobody cares :(
|
||||
}
|
||||
|
||||
public Future<Integer> square(int i) {
|
||||
return Futures.successful(i * i, TypedActor.dispatcher());
|
||||
}
|
||||
|
||||
public Option<Integer> squareNowPlease(int i) {
|
||||
return Option.some(i * i);
|
||||
}
|
||||
|
||||
public int squareNow(int i) {
|
||||
return i * i;
|
||||
}
|
||||
//#typed-actor-impl-methods
|
||||
}
|
||||
//#typed-actor-impl
|
||||
|
||||
@Test public void mustGetTheTypedActorExtension() {
|
||||
|
||||
try {
|
||||
//#typed-actor-extension-tools
|
||||
|
||||
//Returns the Typed Actor Extension
|
||||
TypedActorExtension extension =
|
||||
TypedActor.get(system); //system is an instance of ActorSystem
|
||||
|
||||
//Returns whether the reference is a Typed Actor Proxy or not
|
||||
TypedActor.get(system).isTypedActor(someReference);
|
||||
|
||||
//Returns the backing Akka Actor behind an external Typed Actor Proxy
|
||||
TypedActor.get(system).getActorRefFor(someReference);
|
||||
|
||||
//Returns the current ActorContext,
|
||||
// method only valid within methods of a TypedActor implementation
|
||||
ActorContext context = TypedActor.context();
|
||||
|
||||
//Returns the external proxy of the current Typed Actor,
|
||||
// method only valid within methods of a TypedActor implementation
|
||||
Squarer sq = TypedActor.<Squarer>self();
|
||||
|
||||
//Returns a contextual instance of the Typed Actor Extension
|
||||
//this means that if you create other Typed Actors with this,
|
||||
//they will become children to the current Typed Actor.
|
||||
TypedActor.get(TypedActor.context());
|
||||
|
||||
//#typed-actor-extension-tools
|
||||
} catch (Exception e) {
|
||||
//dun care
|
||||
}
|
||||
}
|
||||
@Test public void createATypedActor() {
|
||||
try {
|
||||
//#typed-actor-create1
|
||||
Squarer mySquarer =
|
||||
TypedActor.get(system).typedActorOf(Squarer.class, SquarerImpl.class, new Props());
|
||||
//#typed-actor-create1
|
||||
//#typed-actor-create2
|
||||
Squarer otherSquarer =
|
||||
TypedActor.get(system).typedActorOf(Squarer.class,
|
||||
new Creator<SquarerImpl>() {
|
||||
public SquarerImpl create() { return new SquarerImpl("foo"); }
|
||||
},
|
||||
new Props(),
|
||||
"name");
|
||||
//#typed-actor-create2
|
||||
|
||||
//#typed-actor-calls
|
||||
//#typed-actor-call-oneway
|
||||
mySquarer.squareDontCare(10);
|
||||
//#typed-actor-call-oneway
|
||||
|
||||
//#typed-actor-call-future
|
||||
Future<Integer> fSquare = mySquarer.square(10); //A Future[Int]
|
||||
//#typed-actor-call-future
|
||||
|
||||
//#typed-actor-call-option
|
||||
Option<Integer> oSquare = mySquarer.squareNowPlease(10); //Option[Int]
|
||||
//#typed-actor-call-option
|
||||
|
||||
//#typed-actor-call-strict
|
||||
int iSquare = mySquarer.squareNow(10); //Int
|
||||
//#typed-actor-call-strict
|
||||
//#typed-actor-calls
|
||||
|
||||
assertEquals(100, Await.result(fSquare, Duration.create(3, TimeUnit.SECONDS)).intValue());
|
||||
|
||||
assertEquals(100, oSquare.get().intValue());
|
||||
|
||||
assertEquals(100, iSquare);
|
||||
|
||||
//#typed-actor-stop
|
||||
TypedActor.get(system).stop(mySquarer);
|
||||
//#typed-actor-stop
|
||||
|
||||
//#typed-actor-poisonpill
|
||||
TypedActor.get(system).poisonPill(otherSquarer);
|
||||
//#typed-actor-poisonpill
|
||||
} catch(Exception e) {
|
||||
//Ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,50 +0,0 @@
|
|||
Guice Integration
|
||||
=================
|
||||
|
||||
All Typed Actors support dependency injection using `Guice <http://code.google.com/p/google-guice/>`_ annotations (such as ‘@Inject’ etc.).
|
||||
The ‘TypedActorManager’ class understands Guice and will do the wiring for you.
|
||||
|
||||
External Guice modules
|
||||
----------------------
|
||||
|
||||
You can also plug in external Guice modules and have not-actors wired up as part of the configuration.
|
||||
Here is an example:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
import static akka.config.Supervision.*;
|
||||
import static akka.config.SupervisorConfig.*;
|
||||
|
||||
TypedActorConfigurator manager = new TypedActorConfigurator();
|
||||
|
||||
manager.configure(
|
||||
new AllForOneStrategy(new Class[]{Exception.class}, 3, 1000),
|
||||
new SuperviseTypedActor[] {
|
||||
new SuperviseTypedActor(
|
||||
Foo.class,
|
||||
FooImpl.class,
|
||||
temporary(),
|
||||
1000),
|
||||
new SuperviseTypedActor(
|
||||
Bar.class,
|
||||
BarImpl.class,
|
||||
permanent(),
|
||||
1000)
|
||||
})
|
||||
.addExternalGuiceModule(new AbstractModule() {
|
||||
protected void configure() {
|
||||
bind(Ext.class).to(ExtImpl.class).in(Scopes.SINGLETON);
|
||||
}})
|
||||
.configure()
|
||||
.inject()
|
||||
.supervise();
|
||||
|
||||
Retrieve the external Guice dependency
|
||||
--------------------------------------
|
||||
|
||||
The external dependency can be retrieved like this:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
Ext ext = manager.getExternalDependency(Ext.class);
|
||||
|
||||
|
|
@ -1,199 +1,166 @@
|
|||
Typed Actors (Java)
|
||||
===================
|
||||
Typed Actors (Scala)
|
||||
====================
|
||||
|
||||
.. sidebar:: Contents
|
||||
|
||||
.. contents:: :local:
|
||||
|
||||
Module stability: **SOLID**
|
||||
Akka Typed Actors is an implementation of the `Active Objects <http://en.wikipedia.org/wiki/Active_object>`_ pattern.
|
||||
Essentially turning method invocations into asynchronous dispatch instead of synchronous that has been the default way since Smalltalk came out.
|
||||
|
||||
The Typed Actors are implemented through `Typed Actors <http://en.wikipedia.org/wiki/Active_object>`_. It uses AOP through `AspectWerkz <http://aspectwerkz.codehaus.org/>`_ to turn regular POJOs into asynchronous non-blocking Actors with semantics of the Actor Model. Each method dispatch is turned into a message that is put on a queue to be processed by the Typed Actor sequentially one by one.
|
||||
Typed Actors consist of 2 "parts", a public interface and an implementation, and if you've done any work in "enterprise" Java, this will be very familiar to you. As with normal Actors you have an external API (the public interface instance) that will delegate methodcalls asynchronously to
|
||||
a private instance of the implementation.
|
||||
|
||||
If you are using the `Spring Framework <http://springsource.org>`_ then take a look at Akka's `Spring integration <spring-integration>`_.
|
||||
The advantage of Typed Actors vs. Actors is that with TypedActors you have a static contract, and don't need to define your own messages, the downside is that it places some limitations on what you can do and what you can't, i.e. you can't use become/unbecome.
|
||||
|
||||
** WARNING: ** Do not configure to use a WorkStealingDispatcher with your TypedActors, it just isn't safe with how TypedActors currently are implemented. This limitation will most likely be removed in the future.
|
||||
Typed Actors are implemented using `JDK Proxies <http://docs.oracle.com/javase/6/docs/api/java/lang/reflect/Proxy.html>`_ which provide a pretty easy-worked API to intercept method calls.
|
||||
|
||||
.. note::
|
||||
|
||||
Just as with regular Akka Untyped Actors, Typed Actors process one call at a time.
|
||||
|
||||
|
||||
The tools of the trade
|
||||
----------------------
|
||||
|
||||
Before we create our first Typed Actor we should first go through the tools that we have at our disposal,
|
||||
it's located in ``akka.actor.TypedActor``.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-extension-tools
|
||||
|
||||
.. warning::
|
||||
|
||||
Same as not exposing ``this`` of an Akka Actor, it's important not to expose ``this`` of a Typed Actor,
|
||||
instead you should pass the external proxy reference, which is obtained from within your Typed Actor as
|
||||
``TypedActor.self()``, this is your external identity, as the ``ActorRef`` is the external identity of
|
||||
an Akka Actor.
|
||||
|
||||
Creating Typed Actors
|
||||
---------------------
|
||||
|
||||
**IMPORTANT:** The Typed Actors class must have access modifier 'public' and can't be a non-static inner class.
|
||||
To create a Typed Actor you need to have one or more interfaces, and one implementation.
|
||||
|
||||
Akka turns POJOs with interface and implementation into asynchronous (Typed) Actors. Akka is using `AspectWerkz’s Proxy <http://blogs.codehaus.org/people/jboner/archives/000914_awproxy_proxy_on_steriods.html>`_ implementation, which is the `most performant <http://docs.codehaus.org/display/AW/AOP+Benchmark>`_ proxy implementation there exists.
|
||||
Our example interface:
|
||||
|
||||
In order to create a Typed Actor you have to subclass the TypedActor base class.
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: imports,typed-actor-iface
|
||||
:exclude: typed-actor-iface-methods
|
||||
|
||||
Here is an example.
|
||||
Our example implementation of that interface:
|
||||
|
||||
If you have a POJO with an interface implementation separation like this:
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: imports,typed-actor-impl
|
||||
:exclude: typed-actor-impl-methods
|
||||
|
||||
.. code-block:: java
|
||||
The most trivial way of creating a Typed Actor instance
|
||||
of our ``Squarer``:
|
||||
|
||||
interface RegistrationService {
|
||||
void register(User user, Credentials cred);
|
||||
User getUserFor(String username);
|
||||
}
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-create1
|
||||
|
||||
.. code-block:: java
|
||||
First type is the type of the proxy, the second type is the type of the implementation.
|
||||
If you need to call a specific constructor you do it like this:
|
||||
|
||||
import akka.actor.TypedActor;
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-create2
|
||||
|
||||
public class RegistrationServiceImpl extends TypedActor implements RegistrationService {
|
||||
public void register(User user, Credentials cred) {
|
||||
... // register user
|
||||
}
|
||||
Since you supply a ``Props``, you can specify which dispatcher to use, what the default timeout should be used and more.
|
||||
Now, our ``Squarer`` doesn't have any methods, so we'd better add those.
|
||||
|
||||
public User getUserFor(String username) {
|
||||
... // fetch user by username
|
||||
return user;
|
||||
}
|
||||
}
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: imports,typed-actor-iface
|
||||
|
||||
Then you can create an Typed Actor out of it by creating it through the 'TypedActor' factory like this:
|
||||
Alright, now we've got some methods we can call, but we need to implement those in ``SquarerImpl``.
|
||||
|
||||
.. code-block:: java
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: imports,typed-actor-impl
|
||||
|
||||
RegistrationService service =
|
||||
(RegistrationService) TypedActor.newInstance(RegistrationService.class, RegistrationServiceImpl.class, 1000);
|
||||
// The last parameter defines the timeout for Future calls
|
||||
Excellent, now we have an interface and an implementation of that interface,
|
||||
and we know how to create a Typed Actor from that, so let's look at calling these methods.
|
||||
|
||||
Creating Typed Actors with non-default constructor
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
Method dispatch semantics
|
||||
-------------------------
|
||||
|
||||
To create a typed actor that takes constructor arguments use a variant of 'newInstance' or 'newRemoteInstance' that takes an instance of a 'TypedActorFactory' in which you can create the TypedActor in any way you like. If you use this method then make sure that no one can get a reference to the actor instance. Touching actor state directly is bypassing the whole actor dispatching mechanism and create race conditions which can lead to corrupt data.
|
||||
Methods returning:
|
||||
|
||||
Here is an example:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
Service service = TypedActor.newInstance(classOf[Service], new TypedActorFactory() {
|
||||
public TypedActor create() {
|
||||
return new ServiceWithConstructorArgsImpl("someString", 500L));
|
||||
});
|
||||
|
||||
Configuration factory class
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Using a configuration object:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
import static java.util.concurrent.TimeUnit.MILLISECONDS;
|
||||
import akka.actor.TypedActorConfiguration;
|
||||
import akka.util.FiniteDuration;
|
||||
|
||||
TypedActorConfiguration config = new TypedActorConfiguration()
|
||||
.timeout(new FiniteDuration(3000, MILLISECONDS));
|
||||
|
||||
RegistrationService service = (RegistrationService) TypedActor.newInstance(RegistrationService.class, config);
|
||||
|
||||
However, often you will not use these factory methods but declaratively define the Typed Actors as part of a supervisor hierarchy. More on that in the :ref:`fault-tolerance-java` section.
|
||||
|
||||
Sending messages
|
||||
----------------
|
||||
|
||||
Messages are sent simply by invoking methods on the POJO, which is proxy to the "real" POJO now. The arguments to the method are bundled up atomically into an message and sent to the receiver (the actual POJO instance).
|
||||
|
||||
One-way message send
|
||||
^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Methods that return void are turned into ‘fire-and-forget’ semantics by asynchronously firing off the message and return immediately. In the example above it would be the 'register' method, so if this method is invoked then it returns immediately:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
// method invocation returns immediately and method is invoke asynchronously using the Actor Model semantics
|
||||
service.register(user, creds);
|
||||
|
||||
Request-reply message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Methods that return something (e.g. non-void methods) are turned into ‘send-and-receive-eventually’ semantics by asynchronously firing off the message and wait on the reply using a Future.
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
// method invocation is asynchronously dispatched using the Actor Model semantics,
|
||||
// but it blocks waiting on a Future to be resolved in the background
|
||||
User user = service.getUser(username);
|
||||
|
||||
Generally it is preferred to use fire-forget messages as much as possible since they will never block, e.g. consume a resource by waiting. But sometimes they are neat to use since they:
|
||||
|
||||
# Simulates standard Java method dispatch, which is more intuitive for most Java developers
|
||||
# Are a neat to model request-reply
|
||||
# Are useful when you need to do things in a defined order
|
||||
|
||||
The same holds for the 'request-reply-with-future' described below.
|
||||
|
||||
Request-reply-with-future message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Methods that return a 'akka.dispatch.Future<TYPE>' are turned into ‘send-and-receive-with-future’ semantics by asynchronously firing off the message and returns immediately with a Future. You need to use the 'future(...)' method in the TypedActor base class to resolve the Future that the client code is waiting on.
|
||||
|
||||
Here is an example:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
public class MathTypedActorImpl extends TypedActor implements MathTypedActor {
|
||||
public Future<Integer> square(int value) {
|
||||
return future(value * value);
|
||||
}
|
||||
}
|
||||
|
||||
MathTypedActor math = TypedActor.typedActorOf(MathTypedActor.class, MathTypedActorImpl.class);
|
||||
|
||||
// This method will return immediately when called, caller should wait on the Future for the result
|
||||
Future<Integer> future = math.square(10);
|
||||
future.await();
|
||||
Integer result = future.get();
|
||||
|
||||
Stopping Typed Actors
|
||||
---------------------
|
||||
|
||||
Once Typed Actors have been created with one of the TypedActor.newInstance methods they need to be stopped with TypedActor.stop to free resources allocated by the created Typed Actor (this is not needed when the Typed Actor is supervised).
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
// Create Typed Actor
|
||||
RegistrationService service = (RegistrationService) TypedActor.newInstance(RegistrationService.class);
|
||||
|
||||
// ...
|
||||
|
||||
// Free Typed Actor resources
|
||||
TypedActor.stop(service);
|
||||
|
||||
When the Typed Actor defines a shutdown callback method (:ref:`fault-tolerance-java`) it will be invoked on TypedActor.stop.
|
||||
|
||||
How to use the TypedActorContext for runtime information access
|
||||
---------------------------------------------------------------
|
||||
|
||||
The 'akka.actor.TypedActorContext' class Holds 'runtime type information' (RTTI) for the Typed Actor. This context is a member field in the TypedActor base class and holds for example the current sender reference, the current sender future etc.
|
||||
|
||||
Here is an example how you can use it to in a 'void' (e.g. fire-forget) method to implement request-reply by using the sender reference:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
class PingImpl implements Ping extends TypedActor {
|
||||
|
||||
public void hit(int count) {
|
||||
Pong pong = (Pong) getContext().getSender();
|
||||
pong.hit(count++);
|
||||
}
|
||||
}
|
||||
|
||||
If the sender, sender future etc. is not available, then these methods will return 'null' so you should have a way of dealing with that scenario.
|
||||
* ``void`` will be dispatched with ``fire-and-forget`` semantics, exactly like ``ActorRef.tell``
|
||||
* ``akka.dispatch.Future<?>`` will use ``send-request-reply`` semantics, exactly like ``ActorRef.ask``
|
||||
* ``scala.Option<?>`` or ``akka.japi.Option<?>`` will use ``send-request-reply`` semantics, but *will* block to wait for an answer,
|
||||
and return None if no answer was produced within the timout, or scala.Some/akka.japi.Some containing the result otherwise.
|
||||
Any exception that was thrown during this call will be rethrown.
|
||||
* Any other type of value will use ``send-request-reply`` semantics, but *will* block to wait for an answer,
|
||||
throwing ``java.util.concurrent.TimeoutException`` if there was a timeout or rethrow any exception that was thrown during this call.
|
||||
|
||||
Messages and immutability
|
||||
-------------------------
|
||||
|
||||
**IMPORTANT**: Messages can be any kind of object but have to be immutable (there is a workaround, see next section). Java or Scala can’t enforce immutability (yet) so this has to be by convention. Primitives like String, int, Long are always immutable. Apart from these you have to create your own immutable objects to send as messages. If you pass on a reference to an instance that is mutable then this instance can be modified concurrently by two different Typed Actors and the Actor model is broken leaving you with NO guarantees and most likely corrupt data.
|
||||
While Akka cannot enforce that the parameters to the methods of your Typed Actors are immutable,
|
||||
we *strongly* recommend that parameters passed are immutable.
|
||||
|
||||
Akka can help you in this regard. It allows you to turn on an option for serializing all messages, e.g. all parameters to the Typed Actor effectively making a deep clone/copy of the parameters. This will make sending mutable messages completely safe. This option is turned on in the :ref:`configuration` file like this:
|
||||
One-way message send
|
||||
^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
.. code-block:: ruby
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-call-oneway
|
||||
|
||||
akka {
|
||||
actor {
|
||||
serialize-messages = on # does a deep clone of messages to ensure immutability
|
||||
}
|
||||
}
|
||||
As simple as that! The method will be executed on another thread; asynchronously.
|
||||
|
||||
This will make a deep clone (using Java serialization) of all parameters.
|
||||
Request-reply message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-call-option
|
||||
|
||||
This will block for as long as the timeout that was set in the ``Props`` of the Typed Actor,
|
||||
if needed. It will return ``None`` if a timeout occurs.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-call-strict
|
||||
|
||||
This will block for as long as the timeout that was set in the ``Props` of the Typed Actor,
|
||||
if needed. It will throw a ``java.util.concurrent.TimeoutException`` if a timeout occurs.
|
||||
|
||||
Request-reply-with-future message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-call-future
|
||||
|
||||
This call is asynchronous, and the Future returned can be used for asynchronous composition.
|
||||
|
||||
Stopping Typed Actors
|
||||
---------------------
|
||||
|
||||
Since Akka's Typed Actors are backed by Akka Actors they must be stopped when they aren't needed anymore.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-stop
|
||||
|
||||
This asynchronously stops the Typed Actor associated with the specified proxy ASAP.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocTestBase.java
|
||||
:include: typed-actor-poisonpill
|
||||
|
||||
This asynchronously stops the Typed Actor associated with the specified proxy
|
||||
after it's done with all calls that were made prior to this call.
|
||||
|
||||
Typed Actor Hierarchies
|
||||
-----------------------
|
||||
|
||||
Since you can obtain a contextual Typed Actor Extension by passing in an ``ActorContext``
|
||||
you can create child Typed Actors by invoking ``typedActorOf(..)`` on that.
|
||||
|
||||
This also works for creating child Typed Actors in regular Akka Actors.
|
||||
|
||||
Lifecycle callbacks
|
||||
-------------------
|
||||
|
||||
By having your Typed Actor implementation class implement any and all of the following:
|
||||
|
||||
* ``TypedActor.PreStart``
|
||||
* ``TypedActor.PostStop``
|
||||
* ``TypedActor.PreRestart``
|
||||
* ``TypedActor.PostRestart``
|
||||
|
||||
You can hook into the lifecycle of your Typed Actor.
|
||||
|
|
|
|||
154
akka-docs/scala/code/akka/docs/actor/TypedActorDocSpec.scala
Normal file
154
akka-docs/scala/code/akka/docs/actor/TypedActorDocSpec.scala
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
package akka.docs.actor
|
||||
|
||||
//#imports
|
||||
import akka.dispatch.{ Promise, Future, Await }
|
||||
import akka.util.duration._
|
||||
import akka.actor.{ ActorContext, TypedActor, Props }
|
||||
|
||||
//#imports
|
||||
|
||||
import org.scalatest.{ BeforeAndAfterAll, WordSpec }
|
||||
import org.scalatest.matchers.MustMatchers
|
||||
import akka.testkit._
|
||||
|
||||
//#typed-actor-iface
|
||||
trait Squarer {
|
||||
//#typed-actor-iface-methods
|
||||
def squareDontCare(i: Int): Unit //fire-forget
|
||||
|
||||
def square(i: Int): Future[Int] //non-blocking send-request-reply
|
||||
|
||||
def squareNowPlease(i: Int): Option[Int] //blocking send-request-reply
|
||||
|
||||
def squareNow(i: Int): Int //blocking send-request-reply
|
||||
//#typed-actor-iface-methods
|
||||
}
|
||||
//#typed-actor-iface
|
||||
|
||||
//#typed-actor-impl
|
||||
class SquarerImpl(val name: String) extends Squarer {
|
||||
|
||||
def this() = this("default")
|
||||
//#typed-actor-impl-methods
|
||||
import TypedActor.dispatcher //So we can create Promises
|
||||
|
||||
def squareDontCare(i: Int): Unit = i * i //Nobody cares :(
|
||||
|
||||
def square(i: Int): Future[Int] = Promise successful i * i
|
||||
|
||||
def squareNowPlease(i: Int): Option[Int] = Some(i * i)
|
||||
|
||||
def squareNow(i: Int): Int = i * i
|
||||
//#typed-actor-impl-methods
|
||||
}
|
||||
//#typed-actor-impl
|
||||
|
||||
//#typed-actor-supercharge
|
||||
trait Foo {
|
||||
def doFoo(times: Int): Unit = println("doFoo(" + times + ")")
|
||||
}
|
||||
|
||||
trait Bar {
|
||||
import TypedActor.dispatcher //So we have an implicit dispatcher for our Promise
|
||||
def doBar(str: String): Future[String] = Promise successful str.toUpperCase
|
||||
}
|
||||
|
||||
class FooBar extends Foo with Bar
|
||||
//#typed-actor-supercharge
|
||||
|
||||
class TypedActorDocSpec extends AkkaSpec(Map("akka.loglevel" -> "INFO")) {
|
||||
|
||||
"get the TypedActor extension" in {
|
||||
val someReference: AnyRef = null
|
||||
|
||||
try {
|
||||
//#typed-actor-extension-tools
|
||||
|
||||
import akka.actor.TypedActor
|
||||
|
||||
//Returns the Typed Actor Extension
|
||||
val extension = TypedActor(system) //system is an instance of ActorSystem
|
||||
|
||||
//Returns whether the reference is a Typed Actor Proxy or not
|
||||
TypedActor(system).isTypedActor(someReference)
|
||||
|
||||
//Returns the backing Akka Actor behind an external Typed Actor Proxy
|
||||
TypedActor(system).getActorRefFor(someReference)
|
||||
|
||||
//Returns the current ActorContext,
|
||||
// method only valid within methods of a TypedActor implementation
|
||||
val c: ActorContext = TypedActor.context
|
||||
|
||||
//Returns the external proxy of the current Typed Actor,
|
||||
// method only valid within methods of a TypedActor implementation
|
||||
val s: Squarer = TypedActor.self[Squarer]
|
||||
|
||||
//Returns a contextual instance of the Typed Actor Extension
|
||||
//this means that if you create other Typed Actors with this,
|
||||
//they will become children to the current Typed Actor.
|
||||
TypedActor(TypedActor.context)
|
||||
|
||||
//#typed-actor-extension-tools
|
||||
} catch {
|
||||
case e: Exception ⇒ //dun care
|
||||
}
|
||||
}
|
||||
|
||||
"create a typed actor" in {
|
||||
//#typed-actor-create1
|
||||
val mySquarer: Squarer =
|
||||
TypedActor(system).typedActorOf[Squarer, SquarerImpl]()
|
||||
//#typed-actor-create1
|
||||
//#typed-actor-create2
|
||||
val otherSquarer: Squarer =
|
||||
TypedActor(system).typedActorOf(classOf[Squarer],
|
||||
new SquarerImpl("foo"),
|
||||
Props(),
|
||||
"name")
|
||||
//#typed-actor-create2
|
||||
|
||||
//#typed-actor-calls
|
||||
//#typed-actor-call-oneway
|
||||
mySquarer.squareDontCare(10)
|
||||
//#typed-actor-call-oneway
|
||||
|
||||
//#typed-actor-call-future
|
||||
val fSquare = mySquarer.square(10) //A Future[Int]
|
||||
//#typed-actor-call-future
|
||||
|
||||
//#typed-actor-call-option
|
||||
val oSquare = mySquarer.squareNowPlease(10) //Option[Int]
|
||||
//#typed-actor-call-option
|
||||
|
||||
//#typed-actor-call-strict
|
||||
val iSquare = mySquarer.squareNow(10) //Int
|
||||
//#typed-actor-call-strict
|
||||
//#typed-actor-calls
|
||||
|
||||
Await.result(fSquare, 3 seconds) must be === 100
|
||||
|
||||
oSquare must be === Some(100)
|
||||
|
||||
iSquare must be === 100
|
||||
|
||||
//#typed-actor-stop
|
||||
TypedActor(system).stop(mySquarer)
|
||||
//#typed-actor-stop
|
||||
|
||||
//#typed-actor-poisonpill
|
||||
TypedActor(system).poisonPill(otherSquarer)
|
||||
//#typed-actor-poisonpill
|
||||
}
|
||||
|
||||
"supercharge" in {
|
||||
//#typed-actor-supercharge-usage
|
||||
val awesomeFooBar = TypedActor(system).typedActorOf[Foo with Bar, FooBar]()
|
||||
|
||||
awesomeFooBar.doFoo(10)
|
||||
val f = awesomeFooBar.doBar("yes")
|
||||
|
||||
TypedActor(system).poisonPill(awesomeFooBar)
|
||||
//#typed-actor-supercharge-usage
|
||||
Await.result(f, 3 seconds) must be === "YES"
|
||||
}
|
||||
}
|
||||
|
|
@ -5,186 +5,171 @@ Typed Actors (Scala)
|
|||
|
||||
.. contents:: :local:
|
||||
|
||||
The Typed Actors are implemented through `Typed Actors <http://en.wikipedia.org/wiki/Active_object>`_. It uses AOP through `AspectWerkz <http://aspectwerkz.codehaus.org/>`_ to turn regular POJOs into asynchronous non-blocking Actors with semantics of the Actor Model. Each method dispatch is turned into a message that is put on a queue to be processed by the Typed Actor sequentially one by one.
|
||||
Akka Typed Actors is an implementation of the `Active Objects <http://en.wikipedia.org/wiki/Active_object>`_ pattern.
|
||||
Essentially turning method invocations into asynchronous dispatch instead of synchronous that has been the default way since Smalltalk came out.
|
||||
|
||||
If you are using the `Spring Framework <http://springsource.org>`_ then take a look at Akka's `Spring integration <spring-integration>`_.
|
||||
Typed Actors consist of 2 "parts", a public interface and an implementation, and if you've done any work in "enterprise" Java, this will be very familiar to you. As with normal Actors you have an external API (the public interface instance) that will delegate methodcalls asynchronously to
|
||||
a private instance of the implementation.
|
||||
|
||||
**WARNING:** Do not configure to use a ``BalancingDispatcher`` with your ``TypedActors``, it just isn't safe with how ``TypedActors`` currently are implemented. This limitation will most likely be removed in the future.
|
||||
The advantage of Typed Actors vs. Actors is that with TypedActors you have a static contract, and don't need to define your own messages, the downside is that it places some limitations on what you can do and what you can't, i.e. you can't use become/unbecome.
|
||||
|
||||
Typed Actors are implemented using `JDK Proxies <http://docs.oracle.com/javase/6/docs/api/java/lang/reflect/Proxy.html>`_ which provide a pretty easy-worked API to intercept method calls.
|
||||
|
||||
.. note::
|
||||
|
||||
Just as with regular Akka Actors, Typed Actors process one call at a time.
|
||||
|
||||
|
||||
The tools of the trade
|
||||
----------------------
|
||||
|
||||
Before we create our first Typed Actor we should first go through the tools that we have at our disposal,
|
||||
it's located in ``akka.actor.TypedActor``.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-extension-tools
|
||||
|
||||
.. warning::
|
||||
|
||||
Same as not exposing ``this`` of an Akka Actor, it's important not to expose ``this`` of a Typed Actor,
|
||||
instead you should pass the external proxy reference, which is obtained from within your Typed Actor as
|
||||
``TypedActor.self``, this is your external identity, as the ``ActorRef`` is the external identity of
|
||||
an Akka Actor.
|
||||
|
||||
Creating Typed Actors
|
||||
---------------------
|
||||
|
||||
**IMPORTANT:** The Typed Actors class must have access modifier 'public' (which is default) and can't be an inner class (unless it is an inner class in an 'object').
|
||||
To create a Typed Actor you need to have one or more interfaces, and one implementation.
|
||||
|
||||
Akka turns POJOs with interface and implementation into asynchronous (Typed) Actors. Akka is using `AspectWerkz’s Proxy <http://blogs.codehaus.org/people/jboner/archives/000914_awproxy_proxy_on_steriods.html>`_ implementation, which is the `most performant <http://docs.codehaus.org/display/AW/AOP+Benchmark>`_ proxy implementation there exists.
|
||||
Our example interface:
|
||||
|
||||
In order to create a Typed Actor you have to subclass the ``TypedActor`` base class.
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: imports,typed-actor-iface
|
||||
:exclude: typed-actor-iface-methods
|
||||
|
||||
Here is an example.
|
||||
Our example implementation of that interface:
|
||||
|
||||
If you have a POJO with an interface implementation separation like this:
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: imports,typed-actor-impl
|
||||
:exclude: typed-actor-impl-methods
|
||||
|
||||
.. code-block:: scala
|
||||
The most trivial way of creating a Typed Actor instance
|
||||
of our Squarer:
|
||||
|
||||
import akka.actor.TypedActor
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-create1
|
||||
|
||||
trait RegistrationService {
|
||||
def register(user: User, cred: Credentials): Unit
|
||||
def getUserFor(username: String): User
|
||||
}
|
||||
First type is the type of the proxy, the second type is the type of the implementation.
|
||||
If you need to call a specific constructor you do it like this:
|
||||
|
||||
.. code-block:: scala
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-create2
|
||||
|
||||
public class RegistrationServiceImpl extends TypedActor with RegistrationService {
|
||||
def register(user: User, cred: Credentials) {
|
||||
... // register user
|
||||
}
|
||||
Since you supply a Props, you can specify which dispatcher to use, what the default timeout should be used and more.
|
||||
Now, our Squarer doesn't have any methods, so we'd better add those.
|
||||
|
||||
def getUserFor(username: String): User = {
|
||||
... // fetch user by username
|
||||
user
|
||||
}
|
||||
}
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: imports,typed-actor-iface
|
||||
|
||||
Then you can create an Typed Actor out of it by creating it through the ``TypedActor`` factory like this:
|
||||
Alright, now we've got some methods we can call, but we need to implement those in SquarerImpl.
|
||||
|
||||
.. code-block:: scala
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: imports,typed-actor-impl
|
||||
|
||||
val service = TypedActor.newInstance(classOf[RegistrationService], classOf[RegistrationServiceImpl], 1000)
|
||||
// The last parameter defines the timeout for Future calls
|
||||
Excellent, now we have an interface and an implementation of that interface,
|
||||
and we know how to create a Typed Actor from that, so let's look at calling these methods.
|
||||
|
||||
Creating Typed Actors with non-default constructor
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
Method dispatch semantics
|
||||
-------------------------
|
||||
|
||||
To create a typed actor that takes constructor arguments use a variant of ``newInstance`` or ``newRemoteInstance`` that takes a call-by-name block in which you can create the Typed Actor in any way you like.
|
||||
Methods returning:
|
||||
|
||||
Here is an example:
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
val service = TypedActor.newInstance(classOf[Service], new ServiceWithConstructorArgs("someString", 500L))
|
||||
|
||||
Configuration factory class
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Using a configuration object:
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
import akka.actor.TypedActorConfiguration
|
||||
import akka.util.Duration
|
||||
import akka.util.duration._
|
||||
|
||||
val config = TypedActorConfiguration()
|
||||
.timeout(3000 millis)
|
||||
|
||||
val service = TypedActor.newInstance(classOf[RegistrationService], classOf[RegistrationServiceImpl], config)
|
||||
|
||||
However, often you will not use these factory methods but declaratively define the Typed Actors as part of a supervisor hierarchy. More on that in the :ref:`fault-tolerance-scala` section.
|
||||
|
||||
Sending messages
|
||||
----------------
|
||||
|
||||
Messages are sent simply by invoking methods on the POJO, which is proxy to the "real" POJO now. The arguments to the method are bundled up atomically into an message and sent to the receiver (the actual POJO instance).
|
||||
|
||||
One-way message send
|
||||
^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Methods that return void are turned into ‘fire-and-forget’ semantics by asynchronously firing off the message and return immediately. In the example above it would be the 'register' method, so if this method is invoked then it returns immediately:
|
||||
|
||||
.. code-block:: java
|
||||
|
||||
// method invocation returns immediately and method is invoke asynchronously using the Actor Model semantics
|
||||
service.register(user, creds)
|
||||
|
||||
Request-reply message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Methods that return something (e.g. non-void methods) are turned into ‘send-and-receive-eventually’ semantics by asynchronously firing off the message and wait on the reply using a Future.
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
// method invocation is asynchronously dispatched using the Actor Model semantics,
|
||||
// but it blocks waiting on a Future to be resolved in the background
|
||||
val user = service.getUser(username)
|
||||
|
||||
Generally it is preferred to use fire-forget messages as much as possible since they will never block, e.g. consume a resource by waiting. But sometimes they are neat to use since they:
|
||||
|
||||
* Simulates standard Java method dispatch, which is more intuitive for most Java developers
|
||||
* Are a neat to model request-reply
|
||||
* Are useful when you need to do things in a defined order
|
||||
|
||||
Request-reply-with-future message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
Methods that return a ``akka.dispatch.Future<TYPE>`` are turned into ‘send-and-receive-with-future’ semantics by asynchronously firing off the message and returns immediately with a Future. You need to use the ``future(...)`` method in the ``TypedActor`` base class to resolve the Future that the client code is waiting on.
|
||||
|
||||
Here is an example:
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
class MathTypedActorImpl extends TypedActor with MathTypedActor {
|
||||
def square(x: Int): Future[Integer] = future(x * x)
|
||||
}
|
||||
|
||||
// create the ping actor
|
||||
val math = TypedActor.newInstance(classOf[MathTyped], classOf[MathTypedImpl])
|
||||
|
||||
// This method will return immediately when called, caller should wait on the Future for the result
|
||||
val future = math.square(10)
|
||||
future.await
|
||||
val result: Int = future.get
|
||||
|
||||
Stopping Typed Actors
|
||||
---------------------
|
||||
|
||||
Once Typed Actors have been created with one of the ``TypedActor.newInstance`` methods they need to be stopped with ``TypedActor.stop`` to free resources allocated by the created Typed Actor (this is not needed when the Typed Actor is supervised).
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
// Create Typed Actor
|
||||
val service = TypedActor.newInstance(classOf[RegistrationService], classOf[RegistrationServiceImpl], 1000)
|
||||
|
||||
// ...
|
||||
|
||||
// Free Typed Actor resources
|
||||
TypedActor.stop(service)
|
||||
|
||||
When the Typed Actor defines a shutdown callback method (:ref:`fault-tolerance-scala`) it will be invoked on ``TypedActor.stop``.
|
||||
|
||||
How to use the TypedActorContext for runtime information access
|
||||
---------------------------------------------------------------
|
||||
|
||||
The ``akka.actor.TypedActorContext`` class Holds 'runtime type information' (RTTI) for the Typed Actor. This context is a member field in the ``TypedActor`` base class and holds for example the current sender reference, the current sender future etc.
|
||||
|
||||
Here is an example how you can use it to in a 'void' (e.g. fire-forget) method to implement request-reply by using the sender reference:
|
||||
|
||||
.. code-block:: scala
|
||||
|
||||
class PingImpl extends TypedActor with Ping {
|
||||
|
||||
def hit(count: Int) {
|
||||
val pong = context.getSender.asInstanceOf[Pong]
|
||||
pong.hit(count++)
|
||||
}
|
||||
}
|
||||
|
||||
If the sender, sender future etc. is not available, then these methods will return ``null`` so you should have a way of dealing with that scenario.
|
||||
* ``Unit`` will be dispatched with ``fire-and-forget`` semantics, exactly like ``ActorRef.tell``
|
||||
* ``akka.dispatch.Future[_]`` will use ``send-request-reply`` semantics, exactly like ``ActorRef.ask``
|
||||
* ``scala.Option[_]`` or ``akka.japi.Option<?>`` will use ``send-request-reply`` semantics, but *will* block to wait for an answer,
|
||||
and return None if no answer was produced within the timout, or scala.Some/akka.japi.Some containing the result otherwise.
|
||||
Any exception that was thrown during this call will be rethrown.
|
||||
* Any other type of value will use ``send-request-reply`` semantics, but *will* block to wait for an answer,
|
||||
throwing ``java.util.concurrent.TimeoutException`` if there was a timeout or rethrow any exception that was thrown during this call.
|
||||
|
||||
Messages and immutability
|
||||
-------------------------
|
||||
|
||||
**IMPORTANT**: Messages can be any kind of object but have to be immutable (there is a workaround, see next section). Java or Scala can’t enforce immutability (yet) so this has to be by convention. Primitives like String, int, Long are always immutable. Apart from these you have to create your own immutable objects to send as messages. If you pass on a reference to an instance that is mutable then this instance can be modified concurrently by two different Typed Actors and the Actor model is broken leaving you with NO guarantees and most likely corrupt data.
|
||||
While Akka cannot enforce that the parameters to the methods of your Typed Actors are immutable,
|
||||
we *strongly* recommend that parameters passed are immutable.
|
||||
|
||||
Akka can help you in this regard. It allows you to turn on an option for serializing all messages, e.g. all parameters to the Typed Actor effectively making a deep clone/copy of the parameters. This will make sending mutable messages completely safe. This option is turned on in the :ref:`configuration` file like this:
|
||||
One-way message send
|
||||
^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
.. code-block:: ruby
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-call-oneway
|
||||
|
||||
akka {
|
||||
actor {
|
||||
# does a deep clone of messages to ensure immutability
|
||||
serialize-messages = on
|
||||
}
|
||||
}
|
||||
As simple as that! The method will be executed on another thread; asynchronously.
|
||||
|
||||
This will make a deep clone (using Java serialization) of all parameters.
|
||||
Request-reply message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-call-option
|
||||
|
||||
This will block for as long as the timeout that was set in the Props of the Typed Actor,
|
||||
if needed. It will return ``None`` if a timeout occurs.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-call-strict
|
||||
|
||||
This will block for as long as the timeout that was set in the Props of the Typed Actor,
|
||||
if needed. It will throw a ``java.util.concurrent.TimeoutException`` if a timeout occurs.
|
||||
|
||||
Request-reply-with-future message send
|
||||
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-call-future
|
||||
|
||||
This call is asynchronous, and the Future returned can be used for asynchronous composition.
|
||||
|
||||
Stopping Typed Actors
|
||||
---------------------
|
||||
|
||||
Since Akkas Typed Actors are backed by Akka Actors they must be stopped when they aren't needed anymore.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-stop
|
||||
|
||||
This asynchronously stops the Typed Actor associated with the specified proxy ASAP.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala
|
||||
:include: typed-actor-poisonpill
|
||||
|
||||
This asynchronously stops the Typed Actor associated with the specified proxy
|
||||
after it's done with all calls that were made prior to this call.
|
||||
|
||||
Typed Actor Hierarchies
|
||||
-----------------------
|
||||
|
||||
Since you can obtain a contextual Typed Actor Extension by passing in an ``ActorContext``
|
||||
you can create child Typed Actors by invoking ``typedActorOf(..)`` on that.
|
||||
|
||||
This also works for creating child Typed Actors in regular Akka Actors.
|
||||
|
||||
Lifecycle callbacks
|
||||
-------------------
|
||||
|
||||
By having your Typed Actor implementation class implement any and all of the following:
|
||||
|
||||
* ``TypedActor.PreStart``
|
||||
* ``TypedActor.PostStop``
|
||||
* ``TypedActor.PreRestart``
|
||||
* ``TypedActor.PostRestart``
|
||||
|
||||
You can hook into the lifecycle of your Typed Actor.
|
||||
|
||||
Supercharging
|
||||
-------------
|
||||
|
||||
Here's an example on how you can use traits to mix in behavior in your Typed Actors.
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala#typed-actor-supercharge
|
||||
|
||||
.. includecode:: code/akka/docs/actor/TypedActorDocSpec.scala#typed-actor-supercharge-usage
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
|
|||
import akka.actor._
|
||||
|
||||
object BeanstalkBasedMailboxExtension extends ExtensionId[BeanstalkMailboxSettings] with ExtensionIdProvider {
|
||||
override def get(system: ActorSystem): BeanstalkMailboxSettings = super.get(system)
|
||||
def lookup() = this
|
||||
def createExtension(system: ActorSystemImpl) = new BeanstalkMailboxSettings(system.settings.config)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
|
|||
import akka.actor._
|
||||
|
||||
object FileBasedMailboxExtension extends ExtensionId[FileBasedMailboxSettings] with ExtensionIdProvider {
|
||||
override def get(system: ActorSystem): FileBasedMailboxSettings = super.get(system)
|
||||
def lookup() = this
|
||||
def createExtension(system: ActorSystemImpl) = new FileBasedMailboxSettings(system.settings.config)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
|
|||
import akka.actor._
|
||||
|
||||
object MongoBasedMailboxExtension extends ExtensionId[MongoBasedMailboxSettings] with ExtensionIdProvider {
|
||||
override def get(system: ActorSystem): MongoBasedMailboxSettings = super.get(system)
|
||||
def lookup() = this
|
||||
def createExtension(system: ActorSystemImpl) = new MongoBasedMailboxSettings(system.settings.config)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import com.typesafe.config.Config
|
|||
import akka.actor._
|
||||
|
||||
object RedisBasedMailboxExtension extends ExtensionId[RedisBasedMailboxSettings] with ExtensionIdProvider {
|
||||
override def get(system: ActorSystem): RedisBasedMailboxSettings = super.get(system)
|
||||
def lookup() = this
|
||||
def createExtension(system: ActorSystemImpl) = new RedisBasedMailboxSettings(system.settings.config)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
|
|||
import akka.actor._
|
||||
|
||||
object ZooKeeperBasedMailboxExtension extends ExtensionId[ZooKeeperBasedMailboxSettings] with ExtensionIdProvider {
|
||||
override def get(system: ActorSystem): ZooKeeperBasedMailboxSettings = super.get(system)
|
||||
def lookup() = this
|
||||
def createExtension(system: ActorSystemImpl) = new ZooKeeperBasedMailboxSettings(system.settings.config)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit.MILLISECONDS
|
|||
import akka.actor.{ ExtensionId, ActorSystem, Extension, ActorSystemImpl }
|
||||
|
||||
object TestKitExtension extends ExtensionId[TestKitSettings] {
|
||||
override def get(system: ActorSystem): TestKitSettings = super.get(system)
|
||||
def createExtension(system: ActorSystemImpl): TestKitSettings = new TestKitSettings(system.settings.config)
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue