Migration Guide 1.3.x to 2.0.x

Creating and starting actors

Actors are created by passing in a Props instance into the actorOf factory method in a ActorRefProvider, which is the ActorSystem or ActorContext. Use the system to create top level actors. Use the context to create actors from other actors. The difference is how the supervisor hierarchy is arranged. When using the context the current actor will be supervisor of the created child actor. When using the system it will be a top level actor, that is supervised by the system (internal guardian actor).

ActorRef.start() has been removed. Actors are now started automatically when created. Remove all invocations of ActorRef.start().

v1.3:

val myActor = Actor.actorOf[MyActor]
myActor.start()

v2.0:

// top level actor
val firstActor = system.actorOf(Props[FirstActor], name = "first")

// child actor
class FirstActor extends Actor {
  val myActor = context.actorOf(Props[MyActor], name = "myactor")

Documentation:

• Actors (Scala)
• Actors (Java)

Stopping actors

ActorRef.stop() has been moved. Use ActorSystem or ActorContext to stop actors.

v1.3:

actorRef.stop()
self.stop()
actorRef ! PoisonPill

v2.0:

context.stop(someChild)
context.stop(self)
system.stop(actorRef)
actorRef ! PoisonPill

Stop all actors

v1.3:

ActorRegistry.shutdownAll()

v2.0:

system.shutdown() // from outside of this system
context.system.shutdown() // from inside any actor

Documentation:

• Actors (Scala)
• Actors (Java)

Identifying Actors

In v1.3 actors have uuid and id field. In v2.0 each actor has a unique logical path.

The ActorRegistry has been replaced by actor paths and lookup with actorFor in ActorRefProvider (ActorSystem or ActorContext). It is no longer possible to obtain references to all actors being implemented by a certain class (the reason being that this property is not known yet when an ActorRef is created because instantiation of the actor itself is asynchronous).

v1.3:

val actor =  Actor.registry.actorFor(uuid)
val actors =  Actor.registry.actorsFor(id)

v2.0:

val actor = context.actorFor("/user/serviceA/aggregator")

Documentation:

• Actor References, Paths and Addresses
• Actors (Scala)
• Actors (Java)

Reply to messages

self.channel has been replaced with unified reply mechanism using sender (Scala) or getSender() (Java). This works for both tell (!) and ask (?). Sending to an actor reference never throws an exception, hence tryTell and tryReply are removed.

v1.3:

self.channel ! result
self.channel tryTell result
self.reply(result)
self.tryReply(result)

v2.0:

sender ! result

Documentation:

• Actors (Scala)
• Actors (Java)

ActorRef.ask()

The mechanism for collecting an actor’s reply in a Future has been reworked for better location transparency: it uses an actor under the hood. This actor needs to be disposable by the garbage collector in case no reply is ever received, and the decision is based upon a timeout. This timeout determines when the actor will stop itself and hence closes the window for a reply to be received; it is independent of the timeout applied when awaiting completion of the Future, however, the actor will complete the Future with an AskTimeoutException when it stops itself.

Since there is no good library default value for the ask-timeout, specification of a timeout is required for all usages as shown below.

Also, since the ask feature is coupling futures and actors, it is no longer offered on the ActorRef itself, but instead as a use pattern to be imported. While Scala’s implicit conversions enable transparent replacement, Java code will have to be changed by more than just adding an import statement.

v1.3:

actorRef ? message // Scala
actorRef.ask(message, timeout); // Java

v2.0 (Scala):

import akka.pattern.ask

implicit val timeout: Timeout = ...
actorRef ? message              // uses implicit timeout
actorRef ask message            // uses implicit timeout
actorRef.ask(message)(timeout)  // uses explicit timeout
ask(actorRef, message)          // uses implicit timeout
ask(actorRef, message)(timeout) // uses explicit timeout

v2.0 (Java):

import akka.pattern.Patterns;

Patterns.ask(actorRef, message, timeout)

Documentation:

• Actors (Scala)
• Actors (Java)

ActorRef.?(msg, timeout)

This method has a dangerous overlap with ActorRef.?(msg)(implicit timeout) due to the fact that Scala allows to pass a Tuple in place of the message without requiring extra parentheses:

actor ? (1, "hallo") // will send a tuple
actor ? (1, Timeout()) // will send 1 with an explicit timeout

To remove this ambiguity, the latter variant is removed in version 2.0. If you were using it before, it will now send tuples where that is not desired. In order to correct all places in the code where this happens, simply import akka.migration.ask instead of akka.pattern.ask to obtain a variant which will give deprecation warnings where the old method signature is used:

import akka.migration.ask

actor ? (1, Timeout(2 seconds)) // will give deprecation warning

UntypedActor.getContext() (Java API only)

getContext() in the Java API for UntypedActor is renamed to getSelf().

v1.3:

actorRef.tell("Hello", getContext());

v2.0:

actorRef.tell("Hello", getSelf());

Documentation:

• Actors (Java)

Configuration

A new, more powerful, configuration utility has been implemented. The format of the configuration file is very similar to the format in v1.3. In addition it also supports configuration files in json and properties format. The syntax is described in the HOCON specification.

v1.3:

include "other.conf"

akka {
  event-handler-level = "DEBUG"
}

v2.0:

include "other"

akka {
  loglevel = "DEBUG"
}

In v1.3 the default name of the configuration file was akka.conf. In v2.0 the default name is application.conf. It is still loaded from classpath or can be specified with java System properties (-D command line arguments).

v1.3:

-Dakka.config=<file path to configuration file>
-Dakka.output.config.source=on

v2.0:

-Dconfig.file=<file path to configuration file>
-Dakka.log-config-on-start=on

Several configuration properties have been changed, such as:

• akka.event-handler-level => akka.loglevel
• dispatcher type values are changed
• akka.actor.throughput => akka.actor.default-dispatcher.throughput
• akka.remote.layer => akka.remote.transport
• the global time-unit property is removed, all durations are specified with duration unit in the property value, timeout = 5s

Verify used configuration properties against the reference Configuration.

Documentation:

• Configuration

Logging

EventHandler API has been replaced by LoggingAdapter, which publish log messages to the event bus. You can still plugin your own actor as event listener with the akka.event-handlers configuration property.

v1.3:

EventHandler.error(exception, this, message)
EventHandler.warning(this, message)
EventHandler.info(this, message)
EventHandler.debug(this, message)
EventHandler.debug(this, "Processing took %s ms".format(duration))

v2.0:

import akka.event.Logging

val log = Logging(context.system, this) // will include system name in message source
val log = Logging(system.eventStream, getClass.getName) // will not include system name
log.error(exception, message)
log.warning(message)
log.info(message)
log.debug(message)
log.debug("Processing took {} ms", duration)

Documentation:

• Logging (Scala)
• Logging (Java)
• Event Bus (Scala)
• Event Bus (Java)

Scheduler

The functionality of the scheduler is identical, but the API is slightly adjusted.

v1.3:

//Schedules to send the "foo"-message to the testActor after 50ms
Scheduler.scheduleOnce(testActor, "foo", 50L, TimeUnit.MILLISECONDS)

// Schedules periodic send of "foo"-message to the testActor after 1s inital delay,
// and then with 200ms between successive sends
Scheduler.schedule(testActor, "foo", 1000L, 200L, TimeUnit.MILLISECONDS)

// Schedules a function to be executed (send the current time) to the testActor after 50ms
Scheduler.scheduleOnce({testActor ! System.currentTimeMillis}, 50L, TimeUnit.MILLISECONDS)

v2.0:

//Schedules to send the "foo"-message to the testActor after 50ms
system.scheduler.scheduleOnce(50 milliseconds, testActor, "foo")

// Schedules periodic send of "foo"-message to the testActor after 1s inital delay,
// and then with 200ms between successive sends
system.scheduler.schedule(1 second, 200 milliseconds, testActor, "foo")

// Schedules a function to be executed (send the current time) to the testActor after 50ms
system.scheduler.scheduleOnce(50 milliseconds) {
  testActor ! System.currentTimeMillis
}

The internal implementation of the scheduler is changed from java.util.concurrent.ScheduledExecutorService to a variant of org.jboss.netty.util.HashedWheelTimer.

Documentation:

• Scheduler (Scala)
• Scheduler (Java)

Supervision

Akka v2.0 implements parental supervision. Actors can only be created by other actors — where the top-level actor is provided by the library — and each created actor is supervised by its parent. In contrast to the special supervision relationship between parent and child, each actor may monitor any other actor for termination.

v1.3:

self.link(actorRef)
self.unlink(actorRef)

v2.0:

class WatchActor extends Actor {
  val actorRef = ...
  // Terminated message will be delivered when the actorRef actor
  // is stopped
  context.watch(actorRef)

  val supervisedChild = context.actorOf(Props[ChildActor])

  def receive = {
    case Terminated(`actorRef`) ⇒ ...
  }
}

Note that link in v1.3 established a supervision relation, which watch doesn't. watch is only a way to get notification, Terminated message, when the monitored actor has been stopped.

Refererence to the supervisor

v1.3:

self.supervisor

v2.0:

context.parent

Supervisor Strategy

v1.3:

val supervisor = Supervisor(
  SupervisorConfig(
    OneForOneStrategy(List(classOf[Exception]), 3, 1000),
    Supervise(
      actorOf[MyActor1],
      Permanent) ::
    Supervise(
      actorOf[MyActor2],
      Permanent) ::
    Nil))

v2.0:

class MyActor extends Actor {
  override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
      case _: ArithmeticException      ⇒ Resume
      case _: NullPointerException     ⇒ Restart
      case _: IllegalArgumentException ⇒ Stop
      case _: Exception                ⇒ Escalate
    }

  def receive = {
    case x =>
  }
}

Documentation:

• Supervision and Monitoring
• Fault Tolerance (Java)
• Fault Tolerance (Scala)
• Actors (Scala)
• Actors (Java)

Dispatchers

Dispatchers are defined in configuration instead of in code.

v1.3:

// in code
val myDispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher(name)
  .withNewThreadPoolWithLinkedBlockingQueueWithCapacity(100)
  .setCorePoolSize(16)
  .setMaxPoolSize(128)
  .setKeepAliveTimeInMillis(60000)
  .build

v2.0:

// in config
my-dispatcher {
  type = Dispatcher
  core-pool-size-factor = 8.0
  max-pool-size-factor  = 16.0
  mailbox-capacity = 100
}

The dispatcher is assigned to the actor in a different way.

v1.3:

actorRef.dispatcher = MyGlobals.myDispatcher
self.dispatcher = MyGlobals.myDispatcher

v2.0:

val myActor = system.actorOf(Props[MyActor].withDispatcher("my-dispatcher"), "myactor")

Documentation:

• Dispatchers (Java)
• Dispatchers (Scala)

Spawn

spawn has been removed and should be replaced by creating a Future. Be careful to not access any shared mutable state closed over by the body.

Scala:

Future { doSomething() } // will be executed asynchronously

Java:

Futures.future<String>(new Callable<String>() {
  public String call() {
    doSomething();
  }
}, executionContext);

Documentation:

• Futures (Scala)
• Futures (Java)
• Akka and the Java Memory Model

HotSwap

In v2.0 become and unbecome metods are located in ActorContext, i.e. context.become and context.unbecome.

The special HotSwap and RevertHotswap messages in v1.3 has been removed. Similar can be implemented with your own message and using context.become and context.unbecome in the actor receiving the message. The rationale is that being able to replace any actor’s behavior generically is not a good idea because actor implementors would have no way to defend against that; hence the change to lay it into the hands of the actor itself.

• Actors (Scala)
• Actors (Java)

Routing

Routing has been redesign with improved performance and additional features as a result.

v1.3:

class MyLoadBalancer extends Actor with LoadBalancer {
  val pinger = actorOf(new Actor { def receive = { case x => println("Pinger: " + x) } }).start()
  val ponger = actorOf(new Actor { def receive = { case x => println("Ponger: " + x) } }).start()

  val seq = new CyclicIterator[ActorRef](List(pinger,ponger))
}
val loadbalancer = actorOf[MyLoadBalancer].start()

v2.0:

val pinger = system.actorOf(Props(new Actor { def receive = { case x ⇒ println("Pinger: " + x) } }))
val ponger = system.actorOf(Props(new Actor { def receive = { case x ⇒ println("Ponger: " + x) } }))
val loadbalancer = system.actorOf(Props().withRouter(RoundRobinRouter(Seq(pinger, ponger))))

Documentation:

• Routing (Scala)
• Routing (Java)

ActorPool

The ActorPool has been replaced by dynamically resizable routers.

v1.3:

class TestPool extends Actor with DefaultActorPool
                             with BoundedCapacityStrategy
                             with ActiveFuturesPressureCapacitor
                             with SmallestMailboxSelector
                             with BasicNoBackoffFilter
{
  def receive = _route
  def lowerBound = 2
  def upperBound = 4
  def rampupRate = 0.1
  def partialFill = true
  def selectionCount = 1
  def instance = actorOf[ExampleActor]
}

v2.0:

// in configuration
akka.actor.deployment {
  /router2 {
    router = round-robin
    resizer {
      lower-bound = 2
      upper-bound = 15
    }
  }
}

// in code
val router2 = system.actorOf(Props[ExampleActor].withRouter(FromConfig())

Documentation:

• Routing (Scala)
• Routing (Java)

Imports

import scala.concurrent.stm._

import scala.concurrent.stm.japi.STM;

import static scala.concurrent.stm.japi.STM.*;

import scala.concurrent.stm.Ref;

Transactions

atomic {
  // do something in transaction
}

atomic { implicit txn =>
  // do something in transaction
}

new Atomic() {
    public Object atomically() {
        // in transaction
        return null;
    }
}.execute();

SomeObject result = new Atomic<SomeObject>() {
    public SomeObject atomically() {
        // in transaction
        return ...;
    }
}.execute();

import static scala.concurrent.stm.japi.STM.atomic;
import java.util.concurrent.Callable;

atomic(new Runnable() {
    public void run() {
        // in transaction
    }
});

SomeObject result = atomic(new Callable<SomeObject>() {
    public SomeObject call() {
        // in transaction
        return ...;
    }
});

Ref

val ref = Ref(0)

val ref = Ref(0)

ref.get // get current value
ref()   // same as get

ref.set(1)  // set to new value, return old value
ref() = 1   // same as set
ref.swap(2) // same as set

ref alter { _ + 1 } // apply a function, return new value

ref.get // get current value
ref()   // same as get

ref.set(1)  // set to new value, return nothing
ref() = 1   // same as set
ref.swap(2) // set and return old value

ref transform { _ + 1 } // apply function, return nothing

ref transformIfDefined { case 1 => 2 } // apply partial function if defined

Ref.View

In v1.3 using a Ref method outside of a transaction would automatically create a single-operation transaction. In v2.0 (in ScalaSTM) there is a Ref.View which provides methods without requiring a current transaction.

ref.single() // returns current value
ref.single() = 1 // set new value

// with atomic this would be:

atomic { implicit t => ref() }
atomic { implicit t => ref() = 1 }

Ref<Integer> ref = new Ref<Integer>(0);

Ref.View<Integer> ref = STM.newRef(0);

ref.get();  // get current value
ref.set(1); // set new value

ref.get();  // get current value
ref.set(1); // set new value

Transaction lifecycle callbacks

atomic {
  deferred {
    // executes when transaction commits
  }
  compensating {
    // executes when transaction aborts
  }
}

atomic { implicit txn =>
  txn.afterCommit { txnStatus =>
    // executes when transaction commits
  }
  txn.afterRollback { txnStatus =>
    // executes when transaction rolls back
  }
}

Transactional Datastructures

In ScalaSTM see TMap, TSet, and TArray for transactional datastructures.

There are helper methods for creating these from Java in japi.STM: newTMap, newTSet, and newTArray. These datastructures implement the scala.collection interfaces and can also be used from Java with Scala's JavaConversions. There are helper methods that apply the conversions, returning java.util Map, Set, and List: newMap, newSet, and newArrayAsList.