Remote Actors (Scala)

Module stability: SOLID

Akka supports starting Actors and Active Objects on remote nodes using a very efficient and scalable NIO implementation built upon JBoss Netty and Google Protocol Buffers .

The usage is completely transparent both in regards to sending messages and error handling and propagation as well as supervision, linking and restarts. You can send references to other Actors as part of the message.

You can find a runnable sample here.

Starting up the remote service

Starting remote service in user code as a library

Here is how to start up the RemoteNode and specify the hostname and port programatically:

import se.scalablesolutions.akka.remote.RemoteNode
 
RemoteNode.start("localhost", 9999)
 
// Specify the classloader to use to load the remote class (actor or active object)
RemoteNode.start("localhost", 9999, classLoader)

Here is how to start up the RemoteNode and specify the hostname and port in the ‘akka.conf’ configuration file (see the section below for details):

import se.scalablesolutions.akka.remote.RemoteNode
 
RemoteNode.start
 
// Specify the classloader to use to load the remote class (actor or active object)
RemoteNode.start(classLoader)

When you start up a 'RemoteNode' then the node is always automatically joining the Akka cluster. If you don't want that then you have to create a 'RemoteServer' intstance and pass in 'false' into the constructor. See below.

Start up more than one remote server

You can create more than one RemoteServer. Just instantiate it with ‘new’ and make sure that the port is unique.

import se.scalablesolutions.akka.remote.RemoteServer
 
val myServer = new RemoteServer
myServer.start("localhost", 9991)
 
val yourServer = new RemoteServer
yourServer.start("localhost", 9992)

If you pass in 'false' into the constructor of the 'RemoteServer' then the server does not join the Akka cluster. Default is that it does join the cluster.

Starting remote service as part of the stand-alone Kernel

You simply need to make sure that the service is turned on in the external ‘akka.conf’ configuration file.

<remote>
  <server>
    service = on
    hostname = "localhost"
    port = 9999
    connection-timeout = 1000 # in millis
  </server>
</remote>

Stopping a RemoteNode or RemoteServer

If you invoke 'shutdown' on the 'RemoteNode' and/or 'RemoteServer' instances then the connection will be closed and the server will be reregistered from the cluster.

RemoteNode.shutdown
 
remoteServer.shutdown

Remote Actors

Akka has two types of remote actors:

• Client-initiated and managed. Here it is the client that creates the remote actor and "moves it" to the server.
• Server-initiated and managed. Here it is the server that creates the remote actor and the client can ask for a handle to this actor.

They are good for different use-cases. The client-initiated are great when you want to monitor an actor on another node since it allows you to link to it and supervise it using the regular supervision semantics. They also make RPC completely transparent. The server-initiated, on the other hand, are great when you have a service running on the server that you want clients to connect to, and you want full control over the actor on the server side for security reasons etc.

Client-managed Remote Actors

When you define an actors as being remote it is instantiated as on the remote host and your local actor becomes a proxy, it works as a handle to the remote actor. The real execution is always happening on the remote node.

Actors can be made remote declaratively by extending the 'RemoteActor(hostname: String, port: Int)' abstract class.

Here is an example:

import se.scalablesolutions.akka.actor.RemoteActor
 
class MyActor extends RemoteActor("192.68.23.769", 9999) {
  def receive = {
    case  "hello" => self.reply("world")
  }
}

Actors can also be made remote after instantiation by invoking one of the ‘makeRemote’ methods on the 'Actor' trait.

def makeRemote(hostname: String, port: Int)
 
def makeRemote(address: InetSocketAddress)

Here is an example:

class MyActor extends Actor {
  self.makeRemote("192.68.23.769", 9999)
 
  def receive = {
    case  "hello" => self.reply("world")
  }
}

An Actor can also start remote child Actors through one of the “spawn/link” methods. These will start, link and make the Actor remote atomically.

...
startLinkRemote(actor, hostname, port)
spawnRemote[MyActor](hostname, port)
spawnLinkRemote[MyActor](hostname, port)
...

Server-managed Remote Actors

Server side setup

The API for server managed remote actors is really simple. 2 methods only:

class HelloWorldActor extends Actor {
  def receive = {
    case "Hello" => self.reply("World")
  }
}
RemoteNode.start("localhost", 9999)
RemoteNode.register("hello-service", actorOf[HelloWorldActor])

Actors created like this are automatically started.

Client side usage

val actor = RemoteClient.actorFor("hello-service", "localhost", 9999)
val result = actor !! "Hello"

There are many variations on the 'RemoteClient#actorFor' method. Here are some of them:

... = RemoteClient.actorFor(className, hostname, port)
... = RemoteClient.actorFor(className, timeout, hostname, port)
... = RemoteClient.actorFor(uuid, className, hostname, port)
... = RemoteClient.actorFor(uuid, className, timeout, hostname, port)
... // etc

All of these also have variations where you can pass in an explicit 'ClassLoader' which can be used when deserializing messages sent from the remote actor.

Running sample

Here is a complete running sample (also available here):

import se.scalablesolutions.akka.actor.Actor
import se.scalablesolutions.akka.remote.{RemoteClient, RemoteNode}
import se.scalablesolutions.akka.util.Logging
 
class HelloWorldActor extends Actor {
  def receive = {
    case "Hello" => self.reply("World")
  }
}
 
object ServerInitiatedRemoteActorServer {
 
  def run = {
    RemoteNode.start("localhost", 9999)
    RemoteNode.register("hello-service", actorOf[HelloWorldActor])
  }
 
  def main(args: Array[String]) = run
}
 
object ServerInitiatedRemoteActorClient extends Logging {
 
  def run = {
    val actor = RemoteClient.actorFor("hello-service", "localhost", 9999)
    val result = actor !! "Hello"
    log.info("Result from Remote Actor: %s", result)
  }
 
  def main(args: Array[String]) = run
}

Connecting and shutting down a RemoteClient explicitly

Normally you should not have to start and stop the 'RemoteClient' explicitly since that is handled by Akka on a demand basis. But if you for some reason want to do that then you can do it like this:

RemoteClient.clientFor(hostname, port).connect // starts and connects the client to the remote server
 
RemoteClient.clientFor(hostname, port).shutdown // disconnects and stops the client

Identifying remote actors

The 'id' field in the 'Actor' class is of importance since it is used as identifier for the remote actor. If you want to create a brand new actor every time you instantiate a remote actor then you have to set the 'id' field to a unique 'String' for each instance. If you want to reuse the same remote actor instance for each new remote actor (of the same class) you create then you don't have to do anything since the 'id' field by default is equal to the name of the actor class.

Here is an example of overriding the 'id' field:

import se.scalablesolutions.akka.util.UUID
 
class MyActor extends RemoteActor("192.68.23.769", 9999) {
  self.id = UUID.newUuid.toString
  def receive = {
    case  "hello" =>  self.reply("world")
  }
}

Remote Active Objects

You can define the Active Object to be a remote service by adding the ‘RemoteAddress’ configuration element in the declarative supervisor configuration:

new Component(
  Foo.class,
  new LifeCycle(new Permanent(), 1000),
  1000,
  new RemoteAddress("localhost", 9999))

You can also define an Active Object to be remote programmatically when creating it explicitly:

ActiveObjectFactory factory = new ActiveObjectFactory();
 
POJO pojo = (POJO) factory.newRemoteInstance(POJO.class, 1000, "localhost", 9999)
 
... // use pojo as usual

Data Compression Configuration

Akka uses compression to minimize the size of the data sent over the wire. Currently it only supports 'zlib' compression but more will come later.

You can configure it like this:

<remote>
  compression-scheme = "zlib" # Options: "zlib" (lzf to come), leave out for no compression
  zlib-compression-level = 6  # Options: 0-9 (1 being fastest and 9 being the most compressed), default is 6
 
  ...
</remote>

Code provisioning

Akka does currently not support automatic code provisioning but requires you to have the remote actor class files available on both the "client" the "server" nodes.
This is something that will be addressed soon. Until then, sorry for the inconvenience.

Remote Client Reconnect Configuration

The Remote Client automatically performs reconnection upon connection failure.

You can configure it like this:

<remote>
  <client>
    reconnect-delay = 5000    # in millis (5 sec default)
    read-timeout = 10000      # in millis (10 sec default)
  <client>
</remote>

Subscribe to Remote Client events

Akka has a subscription API for the 'RemoteClient'. You can register an Actor as a listener and this actor will have to be able to process these events:

sealed trait RemoteClientLifeCycleEvent
case class RemoteClientError(cause: Throwable) extends RemoteClientLifeCycleEvent
case class RemoteClientDisconnected(host: String, port: Int) extends RemoteClientLifeCycleEvent
case class RemoteClientConnected(host: String, port: Int) extends RemoteClientLifeCycleEvent

So a simple listener actor can look like this:

val listener = actor {
  case RemoteClientError(cause) => ... // act upon error
  case RemoteClientDisconnected(hostname, port) => ... // act upon disconnection
  case RemoteClientConnected(hostname, port) => ... // act upon connection
}

Registration and de-registration can be done like this:

RemoteClient.clientFor(hostname, port).register(listener)
...
RemoteClient.clientFor(hostname, port).deregister(listener)

Message Serialization

All messages that are sent to remote actors needs to be serialized to binary format to be able to travel over the wire to the remote node. This is done by letting your messages extend one of the traits in the 'se.scalablesolutions.akka.serialization.Serializable' object. If the messages don't implement any specific serialization trait then the runtime will try to use standard Java serialization.

Here are some examples, but full documentation can be found in the Serialization section.

Scala JSON

case class MyMessage(id: String, value: Tuple2[String, Int]) extends Serializable.ScalaJSON

Protobuf

Protobuf message specification needs to be compiled with 'protoc' compiler.

message ProtobufPOJO {
  required uint64 id = 1;
  required string name = 2;
  required bool status = 3;
}

Using the generated message builder to send the message to a remote actor:

val result = actor !! ProtobufPOJO.newBuilder
    .setId(11)
    .setStatus(true)
    .setName("Coltrane")
    .build

SBinary

case class User(firstNameLastName: Tuple2[String, String], email: String, age: Int) extends Serializable.SBinary[User] {
  import sbinary.DefaultProtocol._
 
  def this() = this(null, null, 0)
 
  implicit object UserFormat extends Format[User] {
    def reads(in : Input) = User(
      read[Tuple2[String, String]](in),
      read[String](in),
      read[Int](in))
    def writes(out: Output, value: User) = {
      write[Tuple2[String, String]](out, value. firstNameLastName)
      write[String](out, value.email)
      write[Int](out, value.age)
    }
  }
 
  def fromBytes(bytes: Array[Byte]) = fromByteArray[User](bytes)
 
  def toBytes: Array[Byte] = toByteArray(this)
}