Actors (Java)

Module stability: SOLID

The Actor Model provides a higher level of abstraction for writing concurrent and distributed systems. It alleviates the developer from having to deal with explicit locking and thread management, making it easier to write correct concurrent and parallel systems. Actors were defined in the 1973 paper by Carl Hewitt but have been popularized by the Erlang language, and used for example at Ericsson with great success to build highly concurrent and reliable telecom systems.

Defining an Actor class

Actors in Java are created either by extending the 'UntypedActor' class and implementing the 'onReceive' method. This method takes two parameters:

• The message as an Object
• A reference to the actor's own UntypedActorRef

The 'UntypedActorRef' is immutable serializable actor reference that you get when you create an UntypedActor. This is the reference that you should use to communicate with the actor, send messages, link to it etc. This reference also functions as the context for the actor and holds run-time type information such as sender of the last message,

Here is an example:

public class SampleUntypedActor extends UntypedActor {
 
  public void onReceive(Object message, UntypedActorRef context) throws Exception {
    if (message instanceof String) log.info("Received String message: %s", message);
    else throw new IllegalArgumentException("Unknown message: " + message);
  }
}

The 'UntypedActor' class inherits from the 'se.scalablesolutions.akka.util.Logging' class which defines a logger in the 'log' field that you can use to log. This logger is configured in the 'akka.conf' configuration file (and is based on the Configgy library which is using Java Logging).

Creating Actors

UntypedActorRef actor = UntypedActor.actorOf(SampleUntypedActor.class);
myActor.start();

Normally you would want to import the 'actorOf' method like this:

import static se.scalablesolutions.akka.actor.UntypedActor.*;
 
UntypedActorRef actor = actorOf(SampleUntypedActor.class);

To avoid prefix it with 'UntypedActor' every time you use it.

You can also start it in the same statement:

UntypedActorRef actor = UntypedActor.actorOf(SampleUntypedActor.class).start();

The call to 'actorOf' returns an instance of 'UntypedActorRef'. This is a handle to the 'UntypedActor' instance which you can use to interact with the Actor, like send messages to it etc. more on this shortly. The 'UntypedActorRef' is immutble and has a one to one relationship with the Actor it represents. The 'UntypedActorRef' is also serializable and network-aware. This means that you can serialize it, send it over the wire and use it on a remote host and it will still be representing the same Actor on the original node, across the network.

Creating Actors with non-default constructor

If your UntypedActor has a constructor that takes parameters then you can't create it using 'actorOf(clazz)'. Instead you can use a variant of 'actorOf' that takes an instance in which you can create the Actor in any way you like. If you use this method then you to make sure that no one can get a reference to the actor instance. If they can get a reference them they can touch state directly in bypass the whole actor dispatching mechanism and create race conditions which can lead to corrupt data.

Here is an example:

UnypedActorRef actor = UntypedActor.actorOf(new MyUntypedActor("service:name", 5)).start();

Identifying Actors

Each UntypedActorRef has two methods:

• context.getUuid();
• context.getId();

The difference is that the 'uuid' is generated by the runtime, guaranteed to be unique and can't be modified. While the 'id' can be set by the user (using 'context.setId(...)', and defaults to Actor class name. You can retrieve Actors by both UUID and ID using the 'ActorRegistry', see the section further down for details.

Messages and immutability

IMPORTANT: Messages can be any kind of object but have to be immutable. Akka can’t enforce immutability (yet) so this has to be by convention.

Send messages

Messages are sent to an Actor through one of the 'send' methods.

• 'sendOneWay' means “fire-and-forget”, e.g. send a message asynchronously and return immediately.
• 'sendRequestReply' means “send-and-reply-eventually”, e.g. send a message asynchronously and wait for a reply through a FutureResult. Here you can specify a timeout. Using timeouts is very important. If no timeout is specified then the actor’s default timeout (set by the 'context.setTimeout(..)' method in the 'UntypedActorRef') is used. This method throws an 'ActorTimeoutException' if the call timed out.
• 'sendRequestReplyFuture' sends a message asynchronously and returns a 'FutureResult'.

In all these methods you have the option of passing along your 'UntypedActorRef' context variable. Make it a practive of doing so because it will allow the receiver actors to be able to respond to your message, since the sender reference is sent along with the message.

Fire-forget

This is the preferred way of sending messages. No blocking waiting for a message. Give best concurrency and scalability characteristics.

actor.sendOneWay("Hello");

Or with the sender reference passed along:

actor.sendOneWay("Hello", context);

If invoked from within an Actor, then the sending actor reference will be implicitly passed along with the message and available to the receiving Actor in its 'context.getSender();' method. He can use this to reply to the original sender or use the 'context.reply(message);' method.

If invoked from an instance that is not an Actor there will be no implicit sender passed along the message and you will get an 'IllegalStateException' if you call 'context.reply(..)'.

Send-And-Receive-Eventually

Using 'sendRequestReply' will send a message to the receiving Actor asynchronously but it will wait for a reply on a 'FutureResult', blocking the sender Actor until either:

• A reply is received, or
• The FutureResult times out and an 'ActorTimeoutException' is thrown.

You can pass an explicit time-out to the 'sendRequestReply' method and if none is specified then the default time-out defined in the sender Actor will be used.

Here are some examples:

UnypedActorRef actorRef = ...
 
try {
  Object result = actorRef.sendRequestReply("Hello", context, 1000);
  ... // handle reply
} catch(ActorTimeoutException e) {
  ... // handle timeout
}

Send-And-Receive-Future

Using 'sendRequestReplyFuture' will send a message to the receiving Actor asynchronously and will immediately return a 'FutureResult'.

FutureResult future= actorRef.sendRequestReplyFuture("Hello", context, 1000);

The 'FutureResult' interface looks like this:

interface FutureResult {
  void await();
  void awaitBlocking();
  boolean isCompleted();
  boolean isExpired();
  long timeoutInNanos();
  Object getResult();
  Throwable getException();
}

So the normal way of working with futures is something like this:

FutureResult future= actorRef.sendRequestReplyFuture("Hello", context, 1000);
future.await();
if (future.isCompleted()) {
  Object result = future.getResult();
  ... // whatever
}

We also have a utility class 'FutureResults' that have a couple of convenience methods:

void awaitAll(Future[] futures);
FutureResult awaitOne(Future[] futures) 

Forward message

You can forward a message from one actor to another. This means that the original sender address/reference is maintained even though the message is going through a 'mediator'. This can be useful when writing actors that work as routers, load-balancers, replicators etc. You need to pass along your UntypedActorRef context variable as well.

context.forward(message, context);

Receive messages

When an actor receives a message is passed into the 'onReceive' method, this is an abstract method on the 'UntypedActor' base class that needs to be defined.

Here is an example:

public class SampleUntypedActor extends UntypedActor {
 
  public void onReceive(Object message, UntypedActorRef context) throws Exception {
    if (message instanceof String) log.info("Received String message: %s", message);
    else throw new IllegalArgumentException("Unknown message: " + message);
  }
}

UntypedActor context

The UntypedActor base class contains almost no member fields or methods to invoke. It only has 'onReceive' message handler and some life-cycle callbacks that you can choose to implement:

Most of the API is in the UnypedActorRef a reference for the actor. As you have seen, this reference is passed into the 'onMessage' method along with the message and it is here, for example, you find methods to reply to messages, send yourself messages, define timeouts, fault tolerance etc., start and stop etc.

Let's start by looking how we can reply to messages in a convenient way using this 'UntypedActorRef' API.

Reply to messages

Reply using the 'replySafe' and 'replyUnsafe' methods

If you want to send a message back to the original sender of the message you just received then you can use the 'context.replyUnsafe(..)' method.

public void onReceive(Object message, UntypedActorRef context) throws Exception {
  if (message instanceof String) {
    String msg = (String)message;
    if (msg.equals("Hello")) {
      // Reply to original sender of message using the 'replyUnsafe' method
      context.replyUnsafe(msg + " from " + context.getUuid());
    }
  }
}

In this case we will a reply back to the Actor that send the message.

The 'replyUnsafe' method throws an 'IllegalStateException' if unable to determine what to reply to, e.g. the sender has not been passed along with the message when invoking one of 'send*' methods. You can also use the more forgiving 'replySafe' method which returns 'true' if reply was sent, and 'false' if unable to determine what to reply to.

public void onReceive(Object message, UntypedActorRef context) throws Exception {
  if (message instanceof String) {
    String msg = (String)message;
    if (msg.equals("Hello")) {
      // Reply to original sender of message using the 'replyUnsafe' method
      if (context.replySafe(msg + " from " + context.getUuid())) ... // success
      else ... // handle failure
    }
  }
}

Reply using the sender reference

If the sender reference (the sender's 'UntypedActorRef') is passed into one ofe the 'send*' methods it will be implicitly passed along together with the message and will be available in the 'Option<UntypedActorRef> getSender()' method on the 'UntypedActorRef. This means that you can use this field to send a message back to the sender.

On this 'Option' you can invoke 'boolean isDefined()' or 'boolean isEmpty()' to check if the sender is available or not, and if it is call 'get()' to get the reference. It's important to know that 'getSender().get()' will throw an exception if there is no sender in scope. The same pattern holds for using the 'getSenderFuture()' in the section below.

public void onReceive(Object message, UntypedActorRef context) throws Exception {
  if (message instanceof String) {
    String msg = (String)message;
    if (msg.equals("Hello")) {
      // Reply to original sender of message using the sender reference
      // also passing along my own refererence (the context)
      if (context.getSender().isDefined) 
        context.getSender().get().sendOneWay(msg + " from " + context.getUuid(), context);
    }
  }
}

Reply using the sender future

If a message was sent with the 'sendRequestReply' or 'sendRequestReplyFuture' methods, which both implements request-reply semantics using Future's, then you either have the option of replying using the 'reply' method as above. This method will then resolve the FutureResult. But you can also get a reference to the FutureResult directly and resolve it yourself or if you would like to store it away to resolve it later, or pass it on to some other Actor to resolve it.

The reference to the Future resides in the 'UntypedActorRef' instance and can be retreived using 'Option<CompletableFuture> getSenderFuture()'.

CompletableFuture is a future with methods for 'completing the future:

• completeWithResult(..)
• completeWithException(..)

Here is an example of how it can be used:

public void onReceive(Object message, UntypedActorRef context) throws Exception {
  if (message instanceof String) {
    String msg = (String)message;
    if (msg.equals("Hello") && context.getSenderFuture().isDefined()) {     
      // Reply to original sender of message using the sender future reference
      context.getSenderFuture().get().completeWithResult(msg + " from " + context.getUuid());
    }
  }
}

Starting actors

Actors are started by invoking the ‘start’ method.

UntypedActorRef actor = UntypedActor.actorOf(SampleUntypedActor.class);
myActor.start();

You can create and start the Actor in a oneliner like this:

UntypedActorRef actor = UntypedActor.actorOf(SampleUntypedActor.class).start();

When you start the actor then it will automatically call the 'init' callback method on the 'UntypedActor'. This is an excellent place to add initialization code for the actor.

@Override 
void init() {
  ... // initialization code
}

Stopping actors

Actors are stopped by invoking the ‘stop’ method.

actor.stop();

When stop is called then a call to the ‘shutdown’ callback method will take place. The Actor can use this callback to implement shutdown behavior.

@Override 
void shutdown() {
  ... // clean up resources
}

You can shut down all Actors in the system by invoking:

ActorRegistry.shutdownAll();

Working with the underlying ActorRef

For some of the API you still need to work with the underlying (Scala) 'ActorRef'. In order to do that you need to know two things:

• The first is how to create a Java friendly 'UntypedActorRef' from an 'ActorRef'. This is done by invoking 'UntypedActorRef.wrap(actorRef)'.
• The second thing is how to get to underlying 'ActorRef' out of the 'UnypedActorRef'. This is done by invoking 'untypedActorRef.actorRef'.

When you know how to do this, then you can work with for example the 'ActorRegistry'.

Killing an Actor

You can kill an actor by sending a 'new Kill()' message. This will restart the actor through regular supervisor semantics.

Use it like this:

// kill the actor called 'victim'
victim.sendOneWay(new Kill());

Actor life-cycle

The actor has a well-defined non-circular life-cycle.

NEW (newly created actor) - can't receive messages (yet)
    => STARTED (when 'start' is invoked) - can receive messages
        => SHUT DOWN (when 'exit' or 'stop' is invoked) - can't do anything