Actors
Describes the XOOM platform Reactive foundation and demonstrates how it is used.
Last updated
Describes the XOOM platform Reactive foundation and demonstrates how it is used.
Last updated
In 1973, Dr. Carl Hewitt and his colleagues formulated the Actor Model. In recent years, the inventor of object orientation, Alan Kay, has stated that the Actor Model retained more of what he thought were the important object ideas. So, when you think of the Actor Model, think of objects done right.
The XOOM Actors toolkit is an implementation of the Actor Model, and all primary platform components are built on our Actor Model implementation. Read on for detailed information on the use of XOOM Actors.
XOOM Actors is an implementation of the Actor Model. The ideas behind the Actor Model are pretty simple, and these points show how XOOM Actors implement it.
The basic unit of computation is expressed through actors. Actors are basically objects, but their behaviors are requested by sending messages asynchronously rather than through directly invoking their methods. This enables all communication between actors to be performed through asynchronous messaging passing. Determined by a scheduler, each actor that has been sent a message will be given the opportunity to receive and process it, which also happens asynchronously.
Actors can create other actors. As Carl Hewitt is known to say, “One actor is no actors. Actors come in systems.” Thus, your applications should use not just some actors, but many actors. Understand that once you start down the road of asynchronous behaviors, you are all in. Just as you don’t kind of go swimming, because you are either completely wet, or you are not, you don't kind of use actors. If you try to fight the asynchrony, you will experience pain and software with very strange bugs.
Each actor can designate the behavior it will exhibit for the next message it receives. This is roughly the State pattern. Any actor using the XOOM Actors toolkit can dynamically become another kind of actor in preparation for handling its current state and any subsequent messages.
What is most unique about the XOOM Actors implementation is type safety by design, and the simplicity with which is it implemented and consumed. All that a programmer needs to understand is interfaces and implementation classes, and they get the asynchrony for free.
Most other actor implementations use a receive method or code block that takes Object (or Any) as a parameter. Thus, the receive needs to determine which messages to accept and which ones are not permitted at any given time; this may be especially necessary when the actor designates its next behavior (it becomes another type of actor). Also what is unique about XOOM Actors is that the current designated behavior can be based on a different interface that is implemented by the actor. In other words, any one actor can implement multiple interfaces and receive messages for any given interface when it chooses to.
Most modern actor model implementations use mailboxes to deliver messages. Each actor has a mailbox where messages are received into a FIFO queue, and each message is processed one at a time on an available thread. This is true for XOOM Actors, yet there are special kinds of mailboxes that have certain advantages (and possibly disadvantages).
You can tune your actor’s world and stage to support any number of threads, but it’s best to limit this number based on the available number of processor hyper-threaded cores, or a bit more. Fundamentally, you can’t run more threads simultaneously than there are available cores, e.g. Runtime.getRuntime().availableProcessors().
Using objects in a typical fashion, such as with Java or C#, we have become accustomed, even addicted, to a blocking paradigm.
Here a Client object invokes a method on a Server object. Understand that this is an in-process (in-VM) invocation, not a remote client and a remote server. The point is, when a method invocation occurs, the Client is blocked until the Server returns from the method invocation. In contrast, the Actor Model works differently.
When the Sender actor wants another actor to provide a service, it sends that actor a message. The message is sent to the Receiver actor and handled asynchronously, but not until a thread is available. The Sender continues moving forward with its current activities, and when completed returns from its own message handling.
As previously stated, with the various Actor Model implementations (e.g. Erlang and Elixir), neither messages or the message receiver are strongly typed. Yet, with XOOM Actors type-safe messages are the fundamental building block, not an experimental afterthought. A strongly-typed Actor Model implementation is important at this time when type safety is in high demand and can provide much more reliable systems.
With its careful but simple design, XOOM Actors are a great foundation on which to build the other tools in the XOOM platform.
Actors collaborate by sending messages, one actor to another. When there are hundreds, thousands, or millions of actors, there are many actors sending messages simultaneously. Still, any one actor can send only one message to one other actor at a time. Following that, the same actor can send a message to the same actor or a different actor. This fulfills the first point of the following description of the actor message-receiving contract.
An actor is a computational entity that, in response to a message it receives, can concurrently:
send a finite number of messages to other actors;
create a finite number of new actors;
designate the behavior to be used for the next message it receives.
There is no assumed sequence to the above actions and they could be carried out in parallel.
Actors also must be able to fulfill the second and third points: actors can create child actors, and actors can prepare themselves for subsequent message receipt.
Now consider a brief tutorial on XOOM Actors. This tutorial takes you through preparing your build environment and also how to implement two actors that collaborate to accomplish a goal.
To get started, create your own playground project to work with. You can name this project playground. If you use Maven, place a dependency into your playground’s pom.xml file.
...
<dependencies>
<dependency>
<groupId>io.vlingo.xoom</groupId>
<artifactId>xoom-actors</artifactId>
<version>x.y.z</version>
</dependency>
</dependencies>
...If you prefer Gradle, insert the following into your build.gradle.
dependencies {
compile 'io.vlingo.xoom:xoom-actors:x.y.z'
}
repositories {
jcenter()
}The x.y.z is a semantic version number and reflects the version of xoom-actors JAR file that you depend on. This number may be something such as 1.8.0. You will find the available versions, including the most recent version, available on one of the supported public repositories.
Additionally, add a JUnit dependency into your build script since the tutorial uses JUnit to run the actor collaboration.
...
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
...Or if using Gradle:
dependencies {
compile 'junit:junit:4.11'
}Although there are a few different ways to configure the XOOM Actors runtime environment, we will skip that step here. It's easier to start with reasonable defaults instead. The configuration approaches are presented later.
Without delay, consider the really fun part—the programming. You are going to create a really basic Ping Pong game.
You need to create a few Java interfaces and classes. There’s a Java interface that acts as the type safe messaging protocol of the first actor that you will create. For now, create a Pinger interface with a single method definition, named ping(), which takes a Ponger as a parameter.
package playground;
import io.vlingo.xoom.actors.Stoppable;
public interface Pinger extends Stoppable {
void ping(final Ponger ponger);
}Next create a Ponger interface the same way, but with a pong() method that takes a Pinger as a parameter.
package playground;
import io.vlingo.xoom.actors.Stoppable;
public interface Ponger extends Stoppable {
void pong(final Pinger pinger);
}Now you have two protocols or two different actors. These define the type-safe behaviors that one or more actors will implement, and the means by which clients will interact with the actors. In case it’s not obvious, Pinger is a client of Ponger, and Ponger is a client of Pinger.
It’s time to create two simple actors. First create one to implement the Pinger protocol.
package playground;
import io.vlingo.xoom.actors.Actor;
public class PingerActor extends Actor implements Pinger {
private final Pinger self;
public PingerActor() {
self = selfAs(Pinger.class);
}
public void ping(final Ponger ponger) {
ponger.pong(self);
}
}After that, create another actor to implement the Ponger protocol.
package playground;
import io.vlingo.xoom.actors.Actor;
public class PongerActor extends Actor implements Ponger {
private final Ponger self;
public PongerActor() {
self = selfAs(Ponger.class);
}
public void pong(final Pinger pinger) {
pinger.ping(self);
}
}You now have two actors that collaborate to play ping pong. The problem is that these actors will play ping pong nonstop, forever, unless we do something to prevent that. Doing so demonstrates how actors can maintain their own state, just like typical objects.
package playground;
import io.vlingo.xoom.actors.Actor;
public class PingerActor extends Actor implements Pinger {
private int count;
private final Pinger self;
public PingerActor() {
count = 0;
self = selfAs(Pinger.class);
}
public void ping(final Ponger ponger) {
if (++count >= 10) {
self.stop();
ponger.stop();
} else {
ponger.pong(self);
}
}
}Looking back at the Pinger and Ponger interface definitions, you will notice that both of these protocols extend the Stoppable protocol. Thus, they can both be stopped by other actors that have a Stoppable reference to them. We use that capability from within PingerActor to cause both actors to stop when the count reaches 10.
Note that in this case the actors are not required to implement their own stop() methods. That’s because the abstract base class, Actor, implements stop() for them. You could override stop() to find out when your actor is being stopped, but that’s not necessarily a good idea. What if you forgot to invoke the super’s stop()? That would make you think that your actor was going to stop, but the actor would never shut down because the Actor base class behavior would never be run. If you want to know when you are being stopped, you can override one of the four life cycle methods instead of stop().
package playground;
import io.vlingo.xoom.actors.Actor;
public class PingerActor extends Actor implements Pinger {
private int count;
private final Pinger self;
public PingerActor() {
count = 0;
self = selfAs(Pinger.class);
}
public void ping(final Ponger ponger) {
if (++count >= 10) {
self.stop();
ponger.stop();
} else {
ponger.pong(self);
}
}
@Override
protected void afterStop() {
logger().log("Pinger " + address() + " just stopped!");
super.afterStop();
}
}All five life cycle methods are:
beforeStart()
afterStop()
beforeRestart(final Throwable reason)
afterRestart(final Throwable reason)
beforeResume(final Throwable reason)
These enable you to see when significant life cycle events occur with your actor. The restart life cycle methods are related to actor supervision. When your actor’s supervisor sees your actor failed with an Exception, it can take a number of actions. Your supervisor can tell your actor to resume, to stop, or to restart. If it tells your actor to resume, the beforeResume() is invoked. When it tells your actor to restart, the beforeRestart() is invoked first, and then the afterRestart() is invoked. Since your actor has failed, it may have been left in an invalid state. In such cases, these three life cycle methods give your actor the opportunity to clean up after the problem that caused the Exception and also reinitialize itself before reacting to its next available protocol message.
The above afterStop() method shows two additional perks of XOOM Actors. All actors have a personal address, which is available through your inherited address() method. Also, all actors have a Logger available via its logger() method. Any information that you log will be output asynchronously through a registered Logger actor, so your actor won't block while file output is performed.
Alright, we have two actors, but how do we bring the actors to life in the first place, and how do we get them to start collaborating in game play? Here’s how you start up the World for your actors to play in.
package playground;
import org.junit.Test;
import io.vlingo.xoom.actors.Definition;
import io.vlingo.xoom.actors.World;
public class PlaygroundTest {
@Test
public void testPlayPingPong() {
final World world = World.startWithDefaults("playground");
final Pinger pinger = world.actorFor(Pinger.class, PingerActor.class);
final Ponger ponger = world.actorFor(Ponger.class, PongerActor.class);
pinger.ping(ponger);
pauseThisThread();
world.terminate();
}
}When this test is run, a World is created. The World is a major component of XOOM Actors. In a nutshell, a World is the primary container within which actors live and play. Generally you would create only one World per service instance. In DDD terms, a World is the root of a Bounded Context. (Don't worry about the use of pauseThisThread(); it is explained below.)
After the World is started, two actors are created, and a reference to their respective protocol is returned. Each actor is created by passing its protocol and the concrete actor type. You may also create actors by means of a Definition. The Definition indicates the class of the actor that implements the protocol, such as PingerActor.class, which implements the Pinger.class protocol. There are four ways to instantiate an actor by means of its constructor:
Design the actor with a zero-parameter constructor, which is the case in the above example.
Pass the implementation class type as the second parameter to actorFor() as seen above, and also pass each constructor parameter following the implementation class type. This has the advantage of making the parameters visible, but they are passed as varargs each of type Object, and are thus not type-safe.
Create and pass a Definition object as the second parameter to actorFor(). The Definition contains the class of the actor implementation and a possibly one or more constructor parameters, or it can pass Definition.NoParameters. See the next code example for how to pass constructor parameters. This approach is also not type safe.
All three of the above actor instantiation options use reflection to call the actor's constructor. Use of reflection can be avoided and at the same time also provide absolute type-safe constructor parameters. To accomplish this, implement a factory for your various actor types using theActorInstantiator, a functional interface included in the xoom-actors SDK and runtime. Construct your specific ActorInstantiator type, pass any actor constructor parameters into the ActorInstantiator constructor. When the actor is ready to be created by the runtime, the ActorInstantiator method blah will be called. At that time call the specific actor's constructor when using new. (See example provided below.)
With these options available, consider the following (non-working) example of the PingerActor taking two parameters, a String and an int, using the Definition approach:
final Pinger pinger = world.actorFor(
Pinger.class,
Definition.has(
PingerActor.class,
Definition.parameters("Hey, yo!", 42),
"name-that-actor"));The simplest way to create an actor with constructor parameters is by means of the following shorthand method, but with the downside that the parameters are not checked for type safety by the compiler but instead at runtime when matching parameter types to a specific constructor:
final Pinger pinger = world.actorFor(Pinger.class, PingerActor.class, "Hey, yo!", 42);The following example employs a type-safe ActorInstantiator, which does not require the use of reflection:
public class PingerInstantiator implements ActorInstantiator<PingerActor> {
private final String message;
private final int value;
public ProtocolInstantiator(final String message, final int value) {
this.message = message;
this.value = value;
}
@Override
public ProtocolActor instantiate() {
return new PingerActor(message, value);
}
@Override
public Class<PingerActor> type() {
return PingerActor.class;
}
}There are three ways to use the PingerInstantiator, each of which is demonstrated separately in the follow example:
// Pass in a Definition
final PingerInstantiator instantiator = new PingerInstantiator("Hey, Yo!", 42);
final Pinger pinger = world.actorFor(Pinger.class, Definition.has(PingerActor.class, instantiator));
// Pass directly as an ActorInstantiator
final Pinger pinger = world.actorFor(Pinger.class, PingerActor.class, new PingerInstantiator("Hey, Yo!", 42));
// Use the FunctionalInterface for lazy instantiation
final Pinger pinger = world.actorFor(Pinger.class, () -> new PingerInstantiator("Hey, Yo!", 42));Look over the XOOM Actors source code repository for the several different ways that an actor can be created, including with a specific parent, a non-default logger, and a specialized supervisor.
One additional point about the unit test is appropriate. As you probably noticed, a method named pauseThisThread() is used.
...
private void pauseThisThread() {
try { Thread.sleep(100); } catch (Exception e) { }
}
...Some sort of coordination is necessary because the actors send and receive all protocol messages asynchronously. Recall that there will be a total of 10 pings. Since the messages are all delivered and reacted to asynchronously, there is no “automatic” way to know when all the messages, including the stop() for both actors, have been delivered.
Don't use Thread.sleep()in your tests or your production services.
Even so, this particular sleep approach is not correct, because on different machines a given sleep time may be insufficient for all messages to process. It's actually guess work to try to get this right. Additionally, if you configure a long-enough sleep time that will work for every possible machine and process load, it's going to make your tests slow on very fast machines and environments. So, one reason for showing you this in the example is to emphasize that you should not use thread sleeps.
The following shows how you can more conveniently test actors without using the thread sleep artifice. It uses the io.vlingo.xoom.actors.testkit.TestUntil component. This example also demonstrates how actors take constructor parameters. In the test method, create an instance of the TestUntil to pass to the PingerActor constructor.
public class PlaygroundTest {
@Test
public void testPlayPingPong() {
final World world = World.start("playground");
final TestUntil until = TestUntil.happenings(1);
final Pinger pinger = world.actorFor(Pinger.class, Definition.has(PingerActor.class, Definition.parameters(until)));
final Ponger ponger = world.actorFor(Ponger.class, Definition.has(PongerActor.class, Definition.NoParameters));
pinger.ping(ponger);
until.completes();
world.terminate();
}
}Then refactor PingerActor to take a TestUntil instance as a constructor parameter.
Additionally, the PingerActor must cause a happened() in its afterStop() method to signal to the test that the Pinger has stopped:
public class PingerActor extends Actor implements Pinger {
private int count;
private final Pinger self;
private final TestUntil until;
public PingerActor(final TestUntil until) {
this.until = until;
this.count = 0;
this.self = selfAs(Pinger.class);
}
...
@Override
protected void afterStop() {
logger().log("Pinger " + address() + " just stopped!");
until.happened();
super.afterStop();
}
}Before the afterStop() method causes the until.happened() the test method will block. As soon as the until.happened() causes its state to transition from 1 to 0, the test will unblock and the World will terminate. This enables the test to complete.
Don't use TestUntil in code that will be used in production. Further, as is seen later in this chapter, you should actually use AccessSafely rather than TestUntil. It is not only thread safe, but also provides memory fences/gates around multi-threaded state modifications during tests.
Although passing a test construct into a production-quality actor is poor design choice, this example is only to show you that there are very reliable ways to test actors in an asynchronous messaging environment. Later you will see much better uses of TestUntil and other io.vlingo.xoom.actors.testkit tools.
In order to make the ping pong playground produce some output, create some log output in the ping() and pong() methods.
// in Pinger
...
public void ping(final Ponger ponger) {
++count;
logger().log("ping " + count);
if (count >= 10) {
self.stop();
ponger.stop();
} else {
ponger.pong(self);
}
}
...
// in Ponger
public void pong(final Pinger pinger) {
logger().log("pong");
pinger.ping(self);
}
...When the test is run, you will see the following output.
vlingo/actors(test): ping 1
vlingo/actors(test): pong
vlingo/actors(test): ping 2
vlingo/actors(test): pong
vlingo/actors(test): ping 3
vlingo/actors(test): pong
vlingo/actors(test): ping 4
vlingo/actors(test): pong
vlingo/actors(test): ping 5
vlingo/actors(test): pong
vlingo/actors(test): ping 6
vlingo/actors(test): pong
vlingo/actors(test): ping 7
vlingo/actors(test): pong
vlingo/actors(test): ping 8
vlingo/actors(test): pong
vlingo/actors(test): ping 9
vlingo/actors(test): pong
vlingo/actors(test): ping 10You can find an implementation of this tutorial code in the xoom-examples repository.
Now that this tutorial has given you some of the most import knowledge about XOOM Actors, you are ready to take a deeper dive into more details about the other facilities provided by XOOM Actors.
This section provides a how-to for the XOOM Actors toolkit API. The details are covered in sections. Some of these details are already demonstrated in the previous sections, including the tutorial.
To start up the XOOM Actors runtime you start the World object as follows.
final World world = World.startWithDefaults("my-world");This starts a World with normal runtime defaults in which Actor instances are created and run. For many uses of XOOM Actors the defaults are the easiest and safest way to use the Actorruntime.
There are a few different ways to start a World. The following is a summary.
Use this API when you want to start a World by loading configurations from the file named xoom-actors.properties. The name is used to name the World instance.
public static World start(final String name)The following API is used when you want to start a World with your own name-value pairs using java.util.Properties defined in code. The details of the xoom-actors.properties file are discussed below.
public static World start(final String name, final java.util.Properties properties)A World can be started using fluent configuration.
public static World start(final String name, final Configuration configuration)The details of programatic Configuration are discussed below.
When you are preparing to shut down your application or service that is using the World, you should use the following to terminate the Actor runtime.
world.terminate();The World::terminate() method is currently a synchronous operation, but in the future will become asynchronous. When the World terminates asynchronously there will be a Completes<T> or callback construct to inform the client when the termination has completed.
The Actor runtime may be configured by means of a file that adheres to the java.util.Properties conventions. Each property is defined by a name followed by = and then a value. For the XOOM Actors toolkit the file must be named xoom-actors.properties and be located in the runtime classpath. The following shows how the standard ConcurrentQueueMailbox can be defined in this properties file.
plugin.name.queueMailbox = true
plugin.queueMailbox.classname =\
io.vlingo.xoom.actors.plugin.mailbox.concurrentqueue.ConcurrentQueueMailboxPlugin
plugin.queueMailbox.defaultMailbox = true
plugin.queueMailbox.numberOfDispatchersFactor = 1.5
plugin.queueMailbox.dispatcherThrottlingCount = 1When defining properties in a properties file, long lines may be continued by placing an escape character of \ at the end of the line to be continued on the next line. To see example properties that can be used, you should review: xoom-actors/src/test/resources/xoom-actors.properties
Also the next subsection shows several configuration type objects and options.
As an alternative to using the file-based configuration, you can instead employ a configuration approach that provides a fluent API, an example of which follows.
final Configuration configuration =
Configuration
.define()
.with(PooledCompletesPluginConfiguration
.define()
.mailbox("queueMailbox")
.poolSize(10))
.with(SharedRingBufferMailboxPluginConfiguration
.define()
.ringSize(65535)
.fixedBackoff(2)
.dispatcherThrottlingCount(10))
.with(ManyToOneConcurrentArrayQueuePluginConfiguration
.define()
.ringSize(65535)
.fixedBackoff(2)
.dispatcherThrottlingCount(10)
.sendRetires(10))
.with(ConcurrentQueueMailboxPluginConfiguration
.define()
.defaultMailbox()
.numberOfDispatchersFactor(1.5f)
.dispatcherThrottlingCount(10))
.with(JDKLoggerPluginConfiguration
.define()
.defaultLogger()
.name("vlingo/actors(test)")
.handlerClass(DefaultHandler.class)
.handlerName("vlingo")
.handlerLevel("ALL"))
.with(CommonSupervisorsPluginConfiguration
.define()
.supervisor("default", "pingSupervisor", Ping.class, PingSupervisorActor.class)
.supervisor("default", "pongSupervisor", Pong.class, PongSupervisorActor.class))
.with(DefaultSupervisorOverridePluginConfiguration
.define()
.supervisor("default", "overrideSupervisor", DefaultSupervisorOverride.class))
.usingMainProxyGeneratedClassesPath("target/classes/")
.usingMainProxyGeneratedSourcesPath("target/generated-sources/")
.usingTestProxyGeneratedClassesPath("target/test-classes/")
.usingTestProxyGeneratedSourcesPath("target/generated-test-sources/");Follow the configuration definition it can be used to start the World instance.
final World world = World.start("my-world", configuration);In addition to the above facilities, a World provides the following. All concrete Actor instance may obtain both their Stage and their World instances as follows, which enables the Actor to reach specific facilities offered by each.
// inside an actor
final Configuration configuration = stage().world().configuration();You may require the use of the means to create unique Actor addresses or a way to produce an address from a primitive or String value. To do so, request it by means of the method addressFactory().
public AddressFactory addressFactory()You may obtain the immutable Configuration of the World runtime. Even if you load your runtime properties from the xoom-actors.properties or your own java.util.Properties definition, all of your runtime configurations are placed in the Configuration.
public Configuration configuration()All messages sent to Actor instances that cannot be delivered for any reasons, such as the Actor instance has previously been stopped, are delivered to the special Actor know as DeadLetters. You may subscribe to receive DeadLetters messages.
public DeadLetters deadLetters()Use the following DeadLetters protocol method to subscribe to its received messages. Your listener Actor must implement the DeadLettersListener protocol.
void registerListener(final DeadLettersListener listener)If you want to obtain the default Logger that is provided to all Actor instances, use the following method.
public Logger defaultLogger()Every top-level application- or service-created Actor is a child of the default parent Actor.
public Actor defaultParent()Every Actor must be assigned to a overarching supervisor. The following provides a reference to the default supervisor of all newly created Actor instance.
public Supervisor defaultSupervisor()Although the default Logger is available through the World interface, there may also be a number of named Logger instances. If you use non-default Logger instances, they may be obtained via the following World facility. All Logger instances obtained through the standard XOOM Actors plugins are backed by actors, and are thus asynchronous by default.
public Logger logger(final String name)The Logger implementation is based on SLF4J. You may configure your Logger to your standards. The following is a simple example that outputs strictly to the console.
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<root level="debug">
<appender-ref ref="STDOUT" />
</root>
</configuration>Another example logs to both the console and a file.
<configuration>
<appender name="FILE" class="ch.qos.logback.core.FileAppender">
<file>myApp.log</file>
<encoder>
<pattern>%date %level [%thread] %logger{10} [%file:%line] %msg%n</pattern>
</encoder>
</appender>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%msg%n</pattern>
</encoder>
</appender>
<root level="debug">
<appender-ref ref="FILE" />
<appender-ref ref="STDOUT" />
</root>
</configuration>There are many references available with far more details about the SLF4J facility. Note that one such configuration supports asynchronous log appending. You may or may not find that this enhances the asynchronous logging that already exists through the logging actor.
The Logger protocol provides several facilities for logging application/service output and exceptions.
void trace(String message)
void trace(String message, Object... args)
void trace(String message, final Throwable throwable)
void debug(String message)
void debug(String message, Object... args)
void debug(String message, final Throwable throwable)
void info(String message)
void info(String message, Object... args)
void info(String message, final Throwable throwable)
void warn(String message)
void warn(String message, Object... args)
void warn(String message, final Throwable throwable)
void error(String message)
void error(String message, Object... args)
void error(String message, final Throwable throwable)The World interface provides some additional facilities, but ones that are useful only for plugins. Those are documented below.
Once you have started a World you are able to create Actor instance to run in it. As a reminder, there is nothing mysterious about actors. Actors are basically objects, but their behaviors are requested by sending messages asynchronously rather than through directly invoking their methods. The following shows you how to create an actor.
final Simple simple = world.actorFor(Simple.class, SimpleActor.class);In this example the World is used to create a new instance of the SimpleActor type. The SimpleActor type implements the protocol defined by the interface named Simple. Thus, actors provide protocol implementations, and they can implement multiple protocols.
In this example the SimpleActor takes no constructor parameters. If it did accept parameters then the parameters could be listed as follows.
final Simple simple = world.actorFor(Simple.class, SimpleActor.class, p1, p2, p3);The parameters must be listed in the order in which to constructor accepts them. In the above example the parameters are listed as p1, p2, and p3. These parameters could be of any type, and each parameter is required to follow the convention that the constructor contract requires.
When the World method actorFor() returns, the requesting client is given a reference to the protocol that provides asynchronous access to the Actor. This reference is used to send messages via methods on the protocol. The Simple protocol is defined as follows.
public interface Simple {
void simpleSay();
}There is a single method named simpleSay(). Messages are sent asynchronously to the concrete Actor instance, which in the case of the above example is an instance of SimpleActor. This behavior is demonstrated by the following expression.
simple.simpleSay();The fundamental behavior of a method invocation on the protocol reference is to reify the method invocation to a message that is sent asynchronously to the Actor. The reification is accomplished by creating a Function that represents the method invocation and enqueuing the Function instance on to the Actor's Mailbox. The Function will later be applied to the Actorinstance when all previously enqueued messages have been processed and a thread becomes available to process the message at the front of the Mailbox. Thus, the method invocation is temporally decoupled from the sending side and the receiving side.
Creating an instance of a concrete Actor type might involve specifying its configuration, such as selecting a non-default mailbox type and setting a non-default Actor name. To do this, use the Definition type:
Simple simple1 =
world.actorFor(Simple.class,
Definition.has(
SimpleActor.class,
Definition.NoParameters,
"arrayQueueMailbox",
"simple-1"));In the above example the Definition is used to pass the SimpleActor type that implements the Simple protocol. Note that this SimpleActor type requires no constructor parameters, as indicated by Definition.NoParameters. The two interesting Definition.has() arguments are "arrayQueueMailbox" and "simple-1". As you likely determined already, the first of the two, "arrayQueueMailbox", is the name of a non-default mailbox type to be used by this SimpleActor instance. The second is the name to be given to the SimpleActor instance.
There are several different predefined Definition.has() overrides. See the Javadoc for the complete set.
There are currently three mailbox types for use by actors.
queueMailbox
A non-blocking unbounded mailbox based on the Java ConcurrentLinkedQueue implementation. This is the default mailbox, unless changed by the user. As the default mailbox it used by all actors unless the actor is specifically created using a different mailbox name. An unbounded mailbox can cause the VM an out-of-memory condition if the actor cannot process messages at least as fast as they are sent.
arrayQueueMailbox
A very fast mailbox based on the Agrona project's non-blocking many-to-one concurrent array queue. The array queue is basically a ring buffer, which means that it can become full and cause incoming messages to be discarded, or at least rejected until space becomes available. Rejecting a message is signaled to the client by the IllegalStateException. Yet, this mailbox will not cause an out-of-memory condition.
ringMailbox
Another fast mailbox based on a non-blocking ring buffer algorithm. The contract of this mailbox is like the other ring buffer implementation, arrayQueueMailbox, but internally is implemented with a different algorithm. However, this mailbox will never reject incoming messages, but this comes at the expense of possible slower enqueuing times. This mailbox will not cause an out-of-memory condition.
For any mailbox that can become full and cause some kind of contention or failure, it can be useful for the owning actor to be a router to some number of fan-out workers. One way to do that is with XOOM Streams and another is with our actor Routing tools.
See the Plugins section for the behavior and configuration details of each mailbox type.
The World does not directly create an Actor. Instead, the World dispatches actorFor() requests to the default Stage. It is the Stage that provides the concrete implementation of actorFor().
A Stage is the abstraction within which Actor instances are maintained. A World has at least one Stage, which is known as the default Stage. Its name is "__defaultStage". If you need to query the instance of the default stage, you use the following World query:
public class World {
...
public Stage stage() {
...
}
...
}Every Stage supports two important behaviors, actorFor() and actorOf(). The actorFor() is a creational method, and there are several overloads supporting various parameter options. Using the method various implementations of the method actorFor() will create a new Actor instance and answer one or more protocols. Once you have a reference to a protocol you are able to send messages to the given Actor that implements the protocol(s).
The actorOf() method has a different purpose. Given an existing Actor that is contained within a given Stage, you may use actorOf() to find the existing Actor by passing the Address of the Actor. The actorOf() answers a Completes<T> because it is an asynchronous operation, answering the requested protocol eventually.
An Actor may obtain its containing Stage and its World as follows.
// inside an actor
final Stage myStage = stage();
final World myWorld = myStage.world();You may create additional Stage instances within the World, but the default Stage is automatically provided when a World starts. The World default Stage is responsible for holding the private root actor and the public root actor. These two instances act as default supervisors, which are responsible for protecting a World from catastrophic failures. Supervision is explained in detail below.
Since the World must create some default operational Actor instances when it is started, it may be best to segregate your application/service Actor instances into another Stage. It's simple to accomplish this. To obtain a reference to an existing Stage by an other name, or to obtain or a newly created Stage by name if non-existing, use one of the two following queries:
public class World {
...
public Stage stageNamed(final String name) {
...
}
public Stage stageNamed(final String name, final Class<? extends Stage> stageType, final AddressFactory addressFactory) {
...
}
...
}If the Stage instance with the given name does not yet exist, a new Stage with that name is created and returned.
We don't suggest creating several or many Stage instances. It's likely that the default Stage and one application/service Stage instance will be sufficient. Yet, we don't set a limit on the number of Stage instances in case more than two would be useful.
Every Stage has its own Schedular, which may be used to produce time-lapsed events that are sent to an Actor.
// inside an actor
final DataPacket packet = new DataPacket(0);
stage().scheduler().schedule(selfAs(Scheduled.class), packet, 100, 1_000);An Actor can schedule itself or another Actor, such as one or more of its children, for some timed event notification. In other words, it need not pass itself as the Scheduled instance.
The above example registers a repeating schedule that will begin within 100 milliseconds of the registration and will repeat every 1 second (1_000 milliseconds). The Actor scheduling itself for notifications must implement the Scheduled protocol, and it passes an Actor enabled reference to that effect using the runtime method selfAs(). The Actor can associate some specific data with which it will be notified on each event, which in this example is the DataPacket instance packet. This DataPacket type is only used for the example, and would be replaced with your own type, or you can pass null if the data is unused.
Similarly you can schedule a single notification. The interval will not not be repeated as in the above example.
// inside an actor
final DataPacket packet = new DataPacket(0);
stage().scheduler().scheduleOnce(selfAs(Scheduled.class), packet, 100, 1_000);When registering a Scheduled object you are provided a Cancellable instance. You may use this instance to cancel one-time or repeating occurrences.
this.cancellable = stage().scheduler().scheduleOnce(...);
...
cancellable.cancel();The Actor receiving the timed event will be notified using the intervalSignal() method of the Scheduled protocol.
public interface Scheduled {
void intervalSignal(final Scheduled scheduled, final Object data);
}It may be implemented something like the following.
@Override
public void intervalSignal(final Scheduled scheduled, final Object data) {
((DataPacket) data).ifAccumulatedEnd(cancellable -> cancellable.cancel());
}Considering Scheduler Latency
When a Scheduler is instantiated it creates a thread pool of 1 thread to manage all scheduling for a single Stage. You can create a unique Scheduler that has its own thread pool in case the Stage's Scheduler is already heavily used. Yet, note that if used as intended, the Scheduled interface registered with the Scheduler is generally implemented by an Actor (as explained above). Thus, in that case, the Scheduler thread is not used to run the ScheduledTask instances. That is, the Scheduler has a thread for tracking all ScheduledTask instances under its care.
From that perspective, it seems unlikely that much latency might be experienced on the Scheduler itself. There seem to be two areas of potential latency that are related to the actor receiving the timed interval messages:
The only latency over and above the Scheduler would be related to how many messages are already in the Actor's mailbox before the Scheduled#intervalSignal() is delivered. The more messages in the mailbox, the more latency to the arrival of the message of concern, intervalSignal().
Another factor to consider is whether the Scheduled#intervalSignal() messages arrive more rapidly than each previous one can be processed. In other words, are the full number of intervalSignal() messages stacking up in the actor's mailbox? If so, consider increasing the time between timer intervals, or finding a way to increase performance of the computing done while handling each message, or both. If the actor's message handling is blocking on I/O you might want to use XOOM Streams to introduce backpressure and a "work stealing" approach to requesting intervals. Additionally, you could employ the Scheduler's scheduleOnce() service and reschedule for one interval each time the actor's processing ends.
Another way to solve latency overhead is by creating a different Mailbox type to increase the number of thread pools for a runtime. You will find this discussion below under Plugins.
Sometimes in applications or microservices require a lot of actors to handle all required compute tasks. This is the case with whole applications or microservices that, for example, have a heavily used domain model where entities are implemented as actors. When this is the case it can be necessary to ensure that the memory heap of any given runtime process is not exhausted. When this is so, the solution runtime can use the Stage's actor eviction service.
To use the eviction service with your solution, you may configure it as a plugin like other plugins. These details are explained later in this chapter.
An Actor is not limited to implementing a single protocol or interface. It could have two or more. Consider again the Scheduler example (above). An Actor implementation that wants to schedule timed event signals will be more than a Scheduled type. Its primary purpose is to be another type of Actor, and to its clients that protocol type is the focus. Consider this example.
public DocumentProcessorActor extends Actor
implements DocumentProcessor, Scheduled {
...
}Here the primary responsibility of the Actor is to be a DocumentProcessor. The fact that it is capable of being scheduled for timer signals is an internal concern only. The clients of this Actor will not know that it can be a Scheduled, only that it is a DocumentProcessor.
// creating a DocumentProcessor
final DocumentProcessor = world.actorFor(
DocumentProcessor.class, DocumentProcessorActor.class, ...);Still, the responsibility of being an effective DocumentProcessor requires it to schedule itself to receives timer signals. The DocumentProcessorActor may be registered with the Scheduler by passing a reference to itself as a Scheduled. You saw this in the above example, but to make it stand out here is another one that isolates the use of the essential selfAs().
// inside an actor
final Schedule scheduled = selfAs(Schedule.class);
final DataPacket packet = new DataPacket(0);
stage().scheduler().scheduleOnce(scheduled, packet, 100, 1_000);All Actor types inherit the method selfAs(Class<T>) to request itself as an Actor that supplies the protocol represented by the generic parameter T. The Actor must implement the T protocol.
Note that you must never pass a reference to yourself using this. Passing this, for example to scheduleOnce(), will cause intervalSignal() notifications to be received as direct method invocations rather than as asynchronous messages. Even though you may think that this could be a good idea, it is definitely a very bad idea. You will eventually experience a race condition in your Actor where a message delivered asynchronously on one thread will collide by accessing data simultaneously with the Scheduler thread that makes the direct method invocations.
An Actor may implement a number of protocols, such as when representing itself as protocols specifically supported by its children or other collaborators. This makes for loose coupling and least knowledge by design. Even so, care should be used to avoid creating an Actor that supports more protocols than is necessary to meet its primary responsibility.
Actors may return values, but must do so by means of Completes<T>, where T is the type of the outcome. If your actor's message protocol that provides a return value answer does not return a Completes<T> value, then the platform runtime will reject the protocol.
The following is a valid protocol for returning a value.
public interface TextToInteger {
Completes<Integer> convertFrom(final String digits);
}Your return value may be provided from inside your actor's message handling method as follows.
public class TextToIntegerActor extends Actor implements TextToInteger {
@Override
public Completes<Integer> convertFrom(final String digits) {
return completes().with(Integer.parseInt(digits));
}
}In some cases an actor is unable to determine a final outcome prior to returning from the message handling method. This is due to the actor depending on another actor requiring asynchronous message sending and an eventual outcome to be answered. Another protocol provides the definition of the Calculator behavior.
public interface Calculator {
Completes<Integer> calculate(final String digits, final int multiplier);
}In such cases the calculate() method may use the Actor base class behavior answerFrom(). This manages the eventual outcome from another actor and and then carries out the final answer.
public class CalculatorActor extends Actor implements Calculator {
private TextToInteger textToInteger;
@Override
public void start() {
textToInteger = childActorFor(TextToInteger.class, Definition.has(TextToIntegerActor.class, Definition.NoParameters));
}
@Override
public Completes<Integer> calculate(final String digits, final int multiplier) {
return answerFrom(textToInteger.convertFrom(digits).andThen(number -> number * multiplier));
}
}The CalculatorActor depends on the previous TextToIntegerActor to convert from a String to an Integer before the CalculatorActor can multiply that number by the given multiplier. Due to this asynchronous behavior, the Actor base class behavior answerFrom() is used to manage the asynchronous messaging and outcome, which is then used to calculate the answer, and provide that answer an eventual outcome to the original client.
A recent account of cascading failure describes tens of thousands of nodes lost during a Kafka failure that caused a Kubernetes cluster to self destruct, taking out an entire infrastructure. Using supervision as bulkheads can save your system from catastrophic failure.
The following diagram illustrates the contrast between how failure is handled in a reactive, Actor Model architecture, and in a typical blocking architecture. It also provides a good indication of why typical blocking architectures can fail catastrophically, and why reactive architectures tend not to.
When an actor throws an exception, or when some synchronous dependency in the actor's use throws an exception, and that exception is caught by the XOOM Actors message delivery component, the supervisor of this actor is informed. The supervisor is responsible for what happens to the actor, such as stopping it or resuming its processing, etc.
The default behavior of the system level supervisors is limited in that they decide what to do about the crashed actor based on static policies. An example might be, "if this actor has crashed 10 times in 5 seconds, I'm going to stop it; otherwise I'll just tell it to resume." Obviously creating your own supervisors provides superior customization, but the default system level supervisors protect again catastrophic failure when no specialized supervisors have been provided by the service/application team.
The Actor Model supports scoped supervision of Actor instances. When the World is created with default configuration, the following hierarchy of supervision is established.
PrivateRootActor (literal)
PublicRootActor (literal)
ApplicationActor1 (example)
ApplicationActor1_1 (example)A concrete Actor that serves as a supervisor must implement the protocol defined by the io.vlingo.xoom.actors.Supervisor. Two of the standard supervisors know as the PrivateRootActor and the PublicRootActor implement the io.vlingo.xoom.actors.Supervisor protocol. Furthermore, if ApplicationActor1 and ApplicationActor1_1 are to serve as supervisors, they too must implement the io.vlingo.xoom.actors.Supervisor protocol.
The above example demonstrates that there is a base supervisor known as the PrivateRootActor. It is the ultimate supervisor that protects the World from catastrophic failure by serving as an impenetrable shield against lower-level failures. Just below the PrivateRootActor is the PublicRootActor. All top-level application/service Actor instances, such as ApplicationActor1 in the above example, are supervised by the PublicRootActor. Any children of the ApplicationActor1, such as ApplicationActor1_1 in the above example, are supervised by ApplicationActor1.
Thus, any newly created concrete Actor whose parent is known and implements the standard Supervisor protocol is the child's supervisor. If the parent is not known or the new Actor is top-level, then both its parent and its supervisor are assigned as the PublicRootActor.
You may override the default supervisor arrangement using the xoom-actors.properties configuration.
plugin.name.override_supervisor = true
plugin.override_supervisor.classname =
io.vlingo.xoom.actors.plugin.supervision.DefaultSupervisorOverridePlugin
plugin.override_supervisor.types =\
[stage=default name=overrideSupervisor
supervisor=io.vlingo.xoom.actors.plugin.supervision.DefaultSupervisorOverride]This will assign the default supervisor as the plugin DefaultSupervisorOverride, which is explained under the plugins section. This override may also be defined using fluent configuration.
final Configuration configuration =
Configuration
.define()
...
.with(DefaultSupervisorOverridePluginConfiguration
.define()
.supervisor("default", "overrideSupervisor",
DefaultSupervisorOverride.class))
...When an Actor throws an exception, the exception is caught by the runtime. When the exception is caught, the Actor is suspended, meaning that it is not permitted to process any messages until the exception has been handled. The exception will be handled by its supervisor when the exception is reified as a message and sent asynchronously to its supervisor. When received, the exception is interpreted and the supervisor's recovery strategy is employed. The supervisor strategy can specify the following actions.
Resume operation of the Actor starting with its next message
Restart the Actor
Stop the Actor, which implies also stopping any of its children
Escalate the recovery up another ancestor level
When the supervisor strategy is to restart, a restart is done only for a maximum number of restarts within a stipulated timeframe. In other words, the Actor may not be permitted to crash repeatedly over an extended period of time. To control restarts or leave them available continuously, the following stipulations may be used.
If the current restart intensity count is within the specified intensity over the given period of time, restart the Actor.
An example of intensity is 10 times within a period of 5 seconds, or 50 times over a period of 1 minute.
You may specify a maximum intensity of an unlimited number of times over an infinite time period.
For examples see the following:
io.vlingo.xoom.actors.PublicRootActor
io.vlingo.xoom.actors.SupervisionStrategy
io.vlingo.xoom.actors.DefaultSupervisor
io.vlingo.xoom.actors.DefaultSupervisorStrategyThe same overrides may be accomplished using the fluent configuration API.
Configuration configuration =
Configuration
.define()
.with(DefaultSupervisorOverridePluginConfiguration
.define()
.supervisor("default", "overrideSupervisor", DefaultSupervisorOverride.class));
There is also a means to provide a supervisor for a given protocol type. For example, the Pinger protocol can be assigned a common supervisor that will provide supervision for all instances of Actors that implement the Pingerprotocol. This can be configured from the xoom-actors.properties file. This example shows how to provide a common supervisor for the Pinger protocol and another for the Ponger protocol.
plugin.name.common_supervisors = true
plugin.common_supervisors.classname = io.vlingo.xoom.actors.plugin.supervision.CommonSupervisorsPlugin
plugin.common_supervisors.types =\
[stage=default name=pingSupervisor protocol=io.vlingo.xoom.actors.supervision.Ping supervisor=io.vlingo.xoom.actors.supervision.PingSupervisorActor]\
[stage=default name=pongSupervisor protocol=io.vlingo.xoom.actors.supervision.Pong supervisor=io.vlingo.xoom.actors.supervision.PongSupervisorActor]Again, this configuration can be made from the programmatic fluent configuration API.
Configuration configuration =
Configuration
.define()
.with(CommonSupervisorsPluginConfiguration
.define()
.supervisor("default", "pingSupervisor", Ping.class, PingSupervisorActor.class)
.supervisor("default", "pongSupervisor", Pong.class, PongSupervisorActor.class));Actor message stowage is a means to temporarily pause message processing for a given actor, but without blocking the thread that initiates the pause. Stowage does this by using a secondary queue to stow all messages that are received until the actor is informed to un-stow messages. At the point of un-stowing, the messages that were stowed will be delivered to the actor before any newly queued messages and until the stowed messages are exhausted; but after messages of any of the stowageOverrides types (see below).
An actor can stow its messages by using the internal (protected) behavior stowMessages() and un-stow messages using disperseStowedMessages(). The Actor protected methods are as follows:
/**
* Starts the process of dispersing any messages stowed for this {@code Actor}.
*/
protected void disperseStowedMessages() {
lifeCycle.environment.mailbox.resume(Mailbox.Paused);
}
/**
* Starts the process of stowing messages for this {@code Actor}
* and registers {@code stowageOverrides} as the protocols that
* will trigger dispersal.
* @param stowageOverrides the {@code Class<T>} array of protocols that will trigger dispersal
*/
protected void stowMessages(final Class<?>... stowageOverrides) {
lifeCycle.environment.mailbox.suspendExceptFor(Mailbox.Paused, stowageOverrides);
}Note that stowMessages() takes the varargs parameter stowageOverrides. This is an array of Class types that, when messages of any of those types are received, will automatically trigger the stowed message dispersal. That is, you enable the mailbox itself to react to a given kind of message that causes un-stowing to happen. When any message of any of the types in the stowageOverrides array is received, it is delivered ahead of any stowed messages. In other words, messages of types in the stowageOverrides array are treated as priority messages.
This is specifically used with Lattice Entity types that auto-persist applied state transition snapshots and/or events to Symbio reactive storage.
The XOOM Actors foundation provides the following plugins.
Completes<T>Some Actor protocols answer outcomes to one or more messages that they handle. Because messages are sent and received asynchronously, the sender will not block until the answered outcome is available. Thus, the contract must be asynchronous for a returned value from theActor to the message-sending client. This contract is satisfied by means of the Completes<T> mechanism. The client sender of an outcome-answering Actor protocol message immediately receives a Completes<T> object. The client then registers a function with the Completes<T> object, and the registered function will be called when the answer becomes available. You may read more about the Completes<T> capabilities in the chapter XOOM Common.
Sometimes an asynchronous message outcome must be helped by some additional Actor-based plumbing. This is where the CompletesEventually plugin is used. This plugin is known as "pooledCompletes", and as its name indicates, a pool of Actor instances is used to process possible high volumes of eventual outcomes.
You may configure this pool using the standard properties file as follows, or by using the fluent configuration API.
plugin.name.pooledCompletes = true
plugin.pooledCompletes.classname =\
io.vlingo.xoom.actors.plugin.completes.PooledCompletesPlugin
plugin.pooledCompletes.pool = 10
plugin.pooledCompletes.mailbox = queueMailboxThis configuration establishes 10 Actor instances that may be used to handle all eventual outcomes. Those Actor instances will use the queueMailbox as their mailbox type (see Mailbox plugin below).
The standard logging capabilities of the io.vlingo.xoom.actors.Logger protocol are provided by the logging plugin found in package io.vlingo.xoom.actors.plugin.logger. The JDKLoggerPlugin is the current default logger found under io.vlingo.xoom.actors.plugin.logger.jdk. There is also a "no op" logger that may be used for quicker testing, packaged under io.vlingo.xoom.actors.plugin.logger.noop.
The configurations for the JDK logger follow those available with java.util.logging. See the following class for configuration details.
io.vlingo.xoom.actors.plugin.logging.jdk.JDKLoggerPlugin.JDKLoggerPluginConfigurationThere are currently three mailbox implementations provided as default plugins, each of which are described in the following content. Configuration examples can be found here.
xoom-actors/src/test/resources/xoom-actors.properties"queueMailbox": A non-blocking unbounded mailbox based on the Java ConcurrentLinkedQueue implementation. This is the default mailbox, unless changed by the user. As the default mailbox it used by all actors unless the actor is specifically created using a different mailbox name.
Being an unbounded mailbox means that sending messages can result in and exception, specifically OutOfMemoryException, if any given Actor receives messages faster than it can process them. Note that this condition will likely require a significant amount of time as messages that can't be processed as quickly as they are queued slowly fill available (e.g. gigabytes of) memory.
Messages sent to an actor that uses this mailbox are ultimately delivered by means of a dispatcher. This "queueMailbox" dispatcher is based on a thread pool executor. You may set the maximum number of threads, or provide a factor (e.g. 0.5, 1.5, or 2.0) to multiply with the total number of processor--hyper thread--to determine the pool size. There are properties used for each option.
plugin.queueMailbox.numberOfDispatchersFactor = 1.5
plugin.queueMailbox.numberOfDispatchers = 0
plugin.queueMailbox.dispatcherThrottlingCount = 1The above numberOfDispatchersFactor example uses a factor of 1.5, which means that the pool of Java threads available for actors using this mailbox will be one-and-a-half times the number of hyper-threads on all cores on the CPU. For example, a 8-core CPU will have 16 total physical hyper-threads, which means that the pool will have 16 x 1.5 or 24 total Java threads.
To set a specific number of Java threads in the pool, set the numberOfDispatchersFactor to 0 and numberOfDispatchers to the exact desired number, such as 50. This would create exactly 50 Java threads in the pool.
There is one additional, dispatcherThrottlingCount. This indicates the maximum number of messages that will be delivered to an actor using this mailbox on each single thread assignment. In other words, if this value is 10, then a single thread assignment to this mailbox could deliver up to 10 total messages before giving up the thread. The most fair configuration is 1, meaning that upon a single thread assignment to a mailbox there would be only 1 message delivered to the actor.
These values may be overridden in the default source code configuration as follows.
final Configuration configuration =
Configuration
.define()
.with(ConcurrentQueueMailboxPluginConfiguration
.define()
.defaultMailbox()
.numberOfDispatchersFactor(0)
.numberOfDispatchers(50)
.dispatcherThrottlingCount(5))
...
final World world = World.start("trading", configuration);The following is the fully-qualified class name of the "queueMailbox".
io.vlingo.xoom.actors.plugin.mailbox.concurrentqueue.ConcurrentQueueMailboxAssuming that the above "queueMailbox" is the default, each of the following two mailboxes ("arrayQueueMailbox" and "ringMailbox") must be explicitly chosen when creating an actor that uses one of these. You can use the following code example to accomplish that, but of course provide the name of the mailbox for you specific choice:
final TradeRouter tradeRouter =
world.actorFor(
TradeRouter.class,
Definition.has(
TradeRouterActor.class,
Definition.parameters(trade),
"arrayQueueMailbox",
"TradeRouter-1"));"arrayQueueMailbox": A very fast mailbox based on the Agrona project's non-blocking many-to-one concurrent array queue.
There are a few available configurations.
plugin.arrayQueueMailbox.size = 65535
plugin.arrayQueueMailbox.fixedBackoff = 2
plugin.arrayQueueMailbox.notifyOnSend = false
plugin.arrayQueueMailbox.dispatcherThrottlingCount = 1
plugin.arrayQueueMailbox.sendRetires = 10The size property determines the total number of individual messages that can fit in the ring buffer at one time. The above shows a limit of 65535. Again note that there are not only 65535 message slots, but also 65535 pre-allocated, reusable message objects.
There is one dedicated thread assigned to the consumer/receiver actor side. Sometimes there will be no messages in the ring yet to be delivered. When this is the case we provide a fixedBackoff property to prevent the Java thread from spinning while waiting for the next message. The above configuration shows a fixed backoff of 2 milliseconds, which means that when there are no available messages to deliver the mailbox thread will sleep for 2 milliseconds. You may increase this value to increase the fixed backoff. You may also set this value to 0 to prevent any backoff; that is, the consumer thread will spin wait in a tight loop. This is the better option if you know that there will be a near constant high throughput of messages sent to the actor.
The notifyOnSend boolean property is used to potentially interrupt the dedicated mailbox thread if it is sleeping. If you are using a high value fixed backoff you may want to consider setting this to true. If you are using fixedBackoff of 0 then set notifyOnSend to false.
The dispatcherThrottlingCount is similar to that for other mailboxes. This indicates the maximum number of messages that will be delivered to an actor using this mailbox during a single dispatch. However, since this mailbox type uses a dedicated thread all that this amounts to is preventing an additional method invocation to have some maximum number of messages delivered rather than just one.
Due to the fact that this mailbox has a limited number of message elements to enqueue, there is a chance that the actor may not be able to process message delivery as fast as its ring buffer is filling. In such cases it is possible that a message could fail enqueuing. Thus, the sendRetires property indicates how many times the sender may attempt to enqueue a message before failing. Note that the margin for retries is a very limited time; only a single for-loop iteration, meaning that even 10 retries is a very small window for the consuming actor to empty one element for the currently delivering message to enqueue. This also doesn't consider that multiple message producers/senders may be simultaneously contending for emptying elements. If the enqueuing fails, the client will be signaled by an IllegalStateException.
You may use programmatic configuration for the "arrayQueueMailbox".
final Configuration configuration =
Configuration
.define()
.with(ManyToOneConcurrentArrayQueuePluginConfiguration
.define()
.ringSize(65535)
.fixedBackoff(0)
.notifyOnSend(false)
.dispatcherThrottlingCount(10)
.sendRetires(10))
...
final World world = World.start("trading", configuration);The following is the fully-qualified class name of the "arrayQueueMailbox".
io.vlingo.xoom.actors.plugin.mailbox.agronampscarrayqueue.ManyToOneConcurrentArrayQueueMailboxAs previously described, this mailbox uses a MPSC data structure, meaning that there may be multiple senders (producers) but only a single receiver (consumer) of the messages. Any number of actors can individually be assigned its own instance of this mailbox.
"ringMailbox": A fast mailbox based on a non-blocking ring buffer algorithm.
Senders may contend to enqueue new message by means of a CAS (compare and set) spin lock. This is slower than the "arrayQueueMailbox" but faster than the "queueMailbox". The primary benefit is that this mailbox will not fail enqueuing to a full ring buffer, but this comes at the expense of possible slower enqueuing times while the actor empties enough slots for all currently sending client contenders.
plugin.ringMailbox.size = 65535
plugin.ringMailbox.fixedBackoff = 2
plugin.ringMailbox.notifyOnSend = false
plugin.ringMailbox.dispatcherThrottlingCount = 1The configuration properties are nearly the same as for the "arrayQueueMailbox", but with a few differences. For example, there is no need for a sendRetires property because an indefinite CAS (compare and set) spin lock is used for retries. Also setting fixedBackoff to 0 has a different affect in the case of this mailbox.
The size property determines the total number of individual messages that can fit in the ring buffer at one time. The above shows a limit of 65535. Again note that there are not only 65535 message slots, but also 65535 pre-allocated, reusable message objects.
There is one dedicated thread assigned to the consumer/receiver actor side. Sometimes there will be no messages in the ring yet to be delivered. When this is the case we provide a fixedBackoff property to prevent the Java thread from spinning while waiting for the next message. The above configuration shows a fixed backoff of 2 milliseconds, which means that when there are no available messages to deliver the mailbox thread will sleep for 2 milliseconds. You may increase this value to increase the fixed backoff. You may also set this value to 0 in order to request capped exponential backoff. Using this backoff approach will double the current backoff in milliseconds until a maximum cap value is reached. The current range is 1 to 4096 milliseconds.
The boolean notifyOnSend property is used to potentially interrupt the dedicated mailbox thread if it is sleeping. If you are using a high value fixed backoff or the capped exponential backoff you should set this the true.
The dispatcherThrottlingCount is similar to that for other mailboxes. This indicates the maximum number of messages that will be delivered to an actor using this mailbox during a single dispatch. However, since this mailbox type uses a dedicated thread all that this amounts to is preventing an additional method invocation to have some maximum number of messages delivered rather than just one.
You may use programmatic configuration for the "ringMailbox".
final Configuration configuration =
Configuration
.define()
.with(SharedRingBufferMailboxPluginConfiguration
.define()
.ringSize(65535)
.fixedBackoff(0)
.notifyOnSend(true)
.dispatcherThrottlingCount(10))
...
final World world = World.start("trading", configuration);The following is the fully-qualified class name of the "ringMailbox".
io.vlingo.xoom.actors.plugin.mailbox.sharedringbuffer.SharedRingBufferMailboxClosely connected with actor mailboxes is the potential need to use multiple thread queues to process different kinds of messages, such as those the require longer to process than most, and that must potentially block such as for I/O to complete. Although there are techniques that can be used to avoid these situations from happening, in some cases it is impractical to do so. In other cases it is simply not possible.
Basically every mailbox type configured as a plugin has its own thread pool. Thus, it is possible to even use the same kind of mailbox but with different configuration names, to force separate thread pools to be created.
In the following example, the xoom-actors.properties is used to explain. Of course you may set up all the configuration programmatically, but here the properties file is used for simplicity:
plugin.name.queueMailbox = true
plugin.queueMailbox.classname = io.vlingo.xoom.actors.plugin.mailbox.concurrentqueue.ConcurrentQueueMailboxPlugin
plugin.queueMailbox.defaultMailbox = true
plugin.queueMailbox.numberOfDispatchersFactor = 1.5
plugin.queueMailbox.numberOfDispatchers = 0
plugin.queueMailbox.dispatcherThrottlingCount = 1
plugin.name.lowLatencyQueueMailbox = true
plugin.lowLatencyQueueMailbox.classname = io.vlingo.xoom.actors.plugin.mailbox.concurrentqueue.ConcurrentQueueMailboxPlugin
plugin.lowLatencyQueueMailbox.defaultMailbox = false
plugin.lowLatencyQueueMailbox.numberOfDispatchersFactor = 0.5
plugin.lowLatencyQueueMailbox.numberOfDispatchers = 0
plugin.lowLatencyQueueMailbox.dispatcherThrottlingCount = 1As seen above, these two mailbox types are the same implementation, but cause the use of different thread pools. The standard and default queueMailbox is allocated with 1.5 times the Java threads as there are physical hyper-threads (not cores, but what Java refers to as processors). If there are 4 cores and 8 physical hyper-threads, this mailbox thread pool will have 12 Java threads.
The second mailbox type is lowLatencyQueueMailbox, which uses the same Mailbox implementation of ConcurrentQueueMailboxPlugin as queueMailbox uses. Even so, lowLatencyQueueMailbox is configured to have only 0.5 times the hyper-threads; if there are 4 cores and 8 hyper-threads, its thread pool will have 4 Java threads. Of course, having a total of 2x the number of Java threads as hyper-threads might stretch things a bit. Yet, we have found that Java can deal with perhaps a few hundred more Java threads than actual hyper-threads. Still, your actual results may vary. It is always most responsible to measure in your own environment.
Note also that you can do the following, for example, to explicitly set the number of threads used by a given mailbox type:
plugin.lowLatencyQueueMailbox.numberOfDispatchersFactor = 0
plugin.lowLatencyQueueMailbox.numberOfDispatchers = 20Even more naturally, you can use a very low-latency mailbox type that far outperforms the ConcurrentQueueMailboxPlugin by something like 3:1 or 4:1 (18-20 million messages per second). This is the mailbox type for which the above mentioned ArrayQueueBenchmark is written against.
plugin.name.arrayQueueMailbox = true
plugin.arrayQueueMailbox.classname = io.vlingo.xoom.actors.plugin.mailbox.agronampscarrayqueue.ManyToOneConcurrentArrayQueuePlugin
plugin.arrayQueueMailbox.defaultMailbox = false
plugin.arrayQueueMailbox.size = 65535
# 0 = exponential back off; >0 = fixed back off
plugin.arrayQueueMailbox.fixedBackoff = 2
plugin.arrayQueueMailbox.notifyOnSend = false
plugin.arrayQueueMailbox.dispatcherThrottlingCount = 1
plugin.arrayQueueMailbox.sendRetires = 10This arrayQueueMailbox is single threaded because it dispatches to only one actor, but multiple producer actors may send messages to the actor that owns it.
We suggest using a variety of approaches to find the one(s) that work best for your domain.
Supervision alternatives may be provided as plugins. The two possible types of extended supervision are as follows.
Name "override_supervisor": Registering this plugin enables an override for the standard default supervisor provided by PublicRootActor.
Name "common_supervisors": Registering this plugin enables any number of different supervisors used to protect against crashed actors of specific protocol types. In other words, you may register a supervisor that will be used to handle exceptions of all actors that implement a specific protocol, and are handling a message for that protocol when the exception is thrown.
The following shows configuration examples for these two kinds of override supervisors found in the properties file.
xoom-actors/src/main/resources/xoom-actors.propertiesplugin.name.override_supervisor = true
plugin.override_supervisor.classname =\
io.vlingo.xoom.actors.plugin.supervision.DefaultSupervisorOverridePlugin
plugin.override_supervisor.types =\
[stage=default name=overrideSupervisor \
supervisor=io.vlingo.xoom.actors.plugin.supervision.DefaultSupervisorOverride]
plugin.name.common_supervisors = true
plugin.common_supervisors.classname =\
io.vlingo.xoom.actors.plugin.supervision.CommonSupervisorsPlugin plugin.common_supervisors.types =\
[stage=default name=pingSupervisor protocol=io.vlingo.xoom.actors.supervision.Ping \
supervisor=io.vlingo.xoom.actors.supervision.PingSupervisorActor] \
[stage=default name=pongSupervisor protocol=io.vlingo.xoom.actors.supervision.Pong \
supervisor=io.vlingo.xoom.actors.supervision.PongSupervisorActor]To configure the eviction service for Stage instances of an application or microservice, use the following properties. These reflect the default values:
plugin.name.directoryEviction = true
plugin.directoryEviction.classname = io.vlingo.xoom.actors.plugin.eviction.DirectoryEvictionPlugin
plugin.directoryEviction.excludedStageNames=__defaultStage
plugin.directoryEviction.enabled = false
plugin.directoryEviction.lruProbeInterval = 40000
plugin.directoryEviction.lruThreshold = 120000
plugin.directoryEviction.fullRatioHighMark = 0.8This configuration indicates that the standard eviction service plugin will be used, which is DirectoryEvictionPlugin. The following properties have specific meaning:
The plugin.directoryEviction.excludedStageNames property offers a comma separated list of names of Stage instances that are not included in eviction services. The default Stage named __defaultStage is never included in eviction services. The default Stage should be safe for relatively low-traffic, quiet actors, such as supervisors, to exist for the life of the service or application and be used only when necessary, which is perhaps seldom. If your Stage name is not listed in this property, it is a candidate for eviction services, but also depends on the value of the following property.
The plugin.directoryEviction.enabled property indicates whether the eviction service is enabled for all Stage instances other than those listed by name in the previous property, where false means disabled by default and true means enabled by default. It might seem counter intuitive that the suggested default is false, or disabled. Most heap pressure will come by way of a growing number of specific application or microservice actors. For example, consider the large number of potential behavioral and data-only entities that could be cached in the XOOM Lattice distributed grid. Those actors that have gone without receiving a message for the longest are considered least recently used (LRU). When heap memory is reaching high pressure (memory is reaching a "highwater mark") then the actors that are least recently used must be detected and stopped, freeing the memory that they consume at runtime. By default, all Grid instances, which are Stage extenders, will be managed by eviction services; that is, unless its name is listed in the above excludedStageNames property.
The plugin.directoryEviction.lruProbeInterval property is the time interval between probes for heap pressure. That is, every interval of this value a scheduler event will fire to cause the heap memory to be probed for a threshold or highwater mark. If the highwater mark is detected, all actors that have reached the LRU threshold will be evicted, as long as they don't currently have any messages waiting for delivery. The default interval is 40 seconds.
The plugin.directoryEviction.lruThreshold property is the length of time considered as the actor LRU threshold, meaning that an actor must be inactive for this time threshold before it is considered for eviction. The default is 2 minutes. Lower this value if heap pressure tends to stay high. Likely two minutes is a low enough interval for solutions that have minimal or average heap pressure. If the solutions tends to require user think time, this timeframe might require increase to reflect the typical user-experience.
The plugin.directoryEviction.fullRatioHighMark property is the level of heap pressure highwater mark that is to be experienced before checking for least recently used actors for eviction. Decrease this value if your heap tends to grow rapidly. The default value is 0.8 or 80% of total maximum heap available. Likely this value would be adjusted with either or both timeframe properties.
To reiterate, for an actor to qualify for eviction, the actor would be least recently used for the lruThreshold and its mailbox must also be empty. That is, the actor will not be evicted if it is just at the cusp of handling a new message. This favors consistency of delivery over memory efficiency.
Every Actor has a life cycle. The Actor is first created and begins life. The Actor may receive and process a number of messages. After some time, the Actor may have reached the extent of its usefulness. At that point some component, possibly another Actor, will request that it be stopped. At that time the Stage containing the Actor instance will remove it from its internal directory, and stop it.
The following protocols are available to Actor instances to help manage its life cycle.
Protocol Startable: All concrete Actor types support this protocol. When the Actor instance is first being started, it receives a start() message. If the concrete Actor wants to react to this message it must override the default behavior, which does nothing.
Protocol Stoppable: All concrete Actor types support this protocol. When the Actor instance is first being stopped, it receives a stop() message. Unlike with the Startable protocol, the base Actor does provide specific critical behavior for its stop() message. Therefore, if the concrete Actor wishes to react to this message by overriding to default, it must ensure that the base implementation is invoked by using super.stop(). If your Actor override attempts to use certain facilities, such as children, the outcome is unpredictable. Due to this somewhat finicky behavior, it may be best for your Actor to instead override the afterStop() life cycle message handler.
Protocol Scheduled: Concrete Actor types by default do not support this protocol, and it must be implemented when using the Scheduler to receive time-based events.
The Actor abstract base class provides five life cycle message handlers:
beforeStart()
afterStop()
beforeResume(final Throwable reason)
beforeRestart(final Throwable reason)
afterRestart(final Throwable reason)
These enable you to see when significant life cycle events occur with your actor.
The beforeStart() message is sent and handled before each Actor is fully started. The default behavior does nothing. If you wish to handle this message in your concrete Actor you must override it.
The afterStop() message is sent and handled after each Actor is fully stopped. The default behavior does nothing. If you wish to handle this message in your concrete Actor you must override it.
The restart life cycle methods are related to actor supervision. When your actor’s supervisor sees your actor failed with an Exception, it can take a number of actions. Your supervisor can tell your actor to resume, to stop, or to restart. If it tells your actor to resume, the beforeResume() is invoked. When it tells your actor to restart, the beforeRestart() is invoked first, and then the afterRestart() is invoked.
Since your actor has failed, it may have been left in an invalid state. In such cases, these life cycle methods give your actor the opportunity to clean up after the problem that caused the Exception and also reinitialize itself before reacting to its next available protocol message.
Since all actors have a Logger available via its logger() method, any information that you log due to a life cycle message will be output asynchronously through a registered Logger. Your Actor won't block while output is performed.
One of the primary capabilities of an Actor is to prepare itself to receive and handle its next message. Since the XOOM Actors foundation fully supports the Actor Model of computation, we enable you to morph any given Actor to the protocol that suites its current state. Thus, an Actor is a state machine and when it reaches any given state its behavior may change to support the specific state.
Changing characters dynamically is managed by the io.vlingo.xoom.actors.Characters mechanism. This test demonstrates how: io.vlingo.xoom.actors.CharactersTest
Testing asynchronous components can be challenging. Yet, the XOOM Actors foundation makes it quite simple to manage. There are a few components that help.
Component TestWorld: This component wraps the standard World object with test capabilities. It supports all of the standard World facilities, but in a specialized way that assists in testing. A TestWorld may be started in exactly the same ways as a World. Creating an Actor through the TestWorld actually creates a TestActor<T>.
Component TestActor<T>: This component is a thin wrapper around the T typed concrete Actor instance. There are three ways to access the Actor to send it a message: (a) Using its T typed protocol, but via a synchronous mailbox; messages are sent and received synchronously. (b) By using the Actor instance directly per the default protocol, enabling direct method invocations. (c) By dynamically casting the Actor instance to another one of its supported protocols, and then directly invoking a method (like b, but using a non-default protocol).
Component TestState: Every Actor implements the TestStateView protocol, enabling tests to request data of the internal state of the Actor. Using method viewTestState() a test can acquire the TestStateof the Actor. A TestState is a set of key-value pairs, providing internal state data that a given concrete Actor is willing to share with a test in order for expectations to be asserted.
Component TestUntil: This component is backed by a Java CountDownLatch, enabling a test to wait on an expected number of happenings in a tested unit. A TestUntil is instantiated by a test to a predetermined number of happenings, such as three. The test invokes until.completes(), which blocks until the number of happenings occurs. The TestUntil instance is shared with a unit under test, and each time that an expected happening takes place, the TestUntil is notified using its happened() protocol. Each happened() notification causes the internal atomic counter to decrement. When the expected number of occurrences has happened, the CountDownLatch reaches zero, and the client unit test side unblocks, allowing the test to complete. This is intended for the most basic tests where data modifications by multiple threads will not cause test issues due to Java memory model unsynchronized updates. See AccessSafely next.
Component AccessSafely: This component uses TestUntil to count down expected steps, but also adds the additional workings to ensure that your test thread sees the same data updates that your actor thread produces. The modern CPU (cores, caching, predictive execution, etc.) are very sophisticated in executing code, but can also produce problems in viewing the same data across different threads. Depending on the design of the language runtime memory model, such as with Java, the data may be synchronized between threads only when certain language facilities are used. When you use AccessSafely to manage the thread-data interplay, you will be able to see data consistently across threads.
You can see these facilities used throughout the XOOM Actors tests. One place to see the test facilities is to view its own tests: io.vlingo.xoom.actors.testkit.TestkitTest
All actors created directly through the TestWorld are assigned a mailbox type of TestMailbox. The TestMailbox does not queue messages, but performs immediate, synchronous delivery. This is done with the purpose of instantly testing the impact of each received message on the receiving actor. Such a test does not have to wait for asynchronous message delivery, making it much simpler to see the resulting message-drive state transition on the receiving actor, for example. When testing actors created through the conventional World, the test must employ a mechanism such as AccessSafely to eventually see and assert against the expected outcomes. It is not always convenient to mock a protocol interface that is backed by a AccessSafely instance, making an actor created through the TestWorld and possessing a TestMailbox an essential tool.
Yet, you must be careful when using an actor created by the TestWorld because such actors don't have asynchronous semantics. If you use several such actors together, it is almost certain that you will experience race conditions. As with all software tools, TestWorld and actors created through it, come with tradeoffs. Use the tools with this knowledge. Note that inside every TestWorld is a normal World that you can query.
World world = testWorld.world();This enables the creation of both TestActor<T> and conventional actor instances.
There is a specific io.vlingo.xoom.actors.testkit tool for maintaining thread-safety of data shared between tests and the XOOM Actors platform components. The tools is named io.vlingo.xoom.actors.testkit.AccessSafely and provides multi-threaded access for both the test thread and the actors being run by the test.
It should be understood that multi-threading is hard to get right, and working with Java on modern CPUs can make it even more complex. Modern CPUs are not only multi-core, but also have performance optimizations that perform what may seem to be counter intuitive operations. One such operations is executing code out of sequence from the actual implementation seen in source code. Thus, even when using AccessSafely to protect access to data until it has been written to by an actor running on a separate thread, if the visible/logical program code is not run in the expected order, other threads can see unexpected values.
atomicValue.set(1);
until.completed();In this very simple example, what would happen if the CPU decided that it would be more performant to execute the second statement first, and the first statement second? Impossible? Consider these statements in the Wikipedia article on the Java Memory Model.
The major caveat of this is that as-if-serial semantics do not prevent different threads from having different views of the data.... The basic rules imply that individual actions can be reordered, as long as the as-if-serial semantics of the thread are not violated.
What would happen is the test thread that is blocking on the TestUntil completion would unblock and attempt to read the atomicValue. If the other thread has not yet written 1 to the atomicValue, the test thread will see the wrong value and its assertion will fail. When you review the code you will insist that it is impossible for the atomicValue to remain unchanged if the TestUntil completed, and yet due to the CPU architecture you would be wrong.
Next, consider a related and critical point from the same Java Memory Model article.
Actions that imply communication between threads, such as the acquisition or release of a lock, ensure that actions that happen prior to them are seen by other threads that see their effects. For example, everything that happens before the release of a lock will be seen to be ordered before and visible to everything that happens after a subsequent acquisition of that same lock.
The AccessSafely testkit component helps you avoid such problems and solve them using the aforementioned techniques, relieving you from the burden of writing complex test code.
To use this tool, first create a component to be used under the control of the test, such as a mock object, such as one passed to an actor. The mock would resemble the expected asynchronous argument, but is not an asynchronous implementation. One such example is creating a mock implementor of PersistResultInterest, such as is done in XOOM Symbio with MockPersistResultInterest. Please see the tests that use the mocks. The following is a snippet from one part of the mock.
Inside the MockPersistResultInterest you set up some writers for the inside to write with, and readers for the test to read with.
public AccessSafely afterCompleting(final int times) {
access = AccessSafely
.afterCompleting(times)
.writingWith("add", (value) -> persistentObjects.add((int) value))
.writingWith("addAll", (values) -> persistentObjects.addAll((List) values))
.readingWith("object", (index) -> persistentObjects.get((int) index))
.readingWith("size", () -> persistentObjects.size());
return access;
}This creates a new AccessSafely for some number of write occurrences followed by the registration of properties with write and read access. The property names are provided as String values. The actual writers and readers are lambdas.
The times of predetermined actions is backed by TestUntil inside (see above). When you first readFrom("some-property") you will block until those actions have been completed, at which time you will get the return value of the property that you asked for. In the above example there are four properties including "size" and "object". When the test client uses readFrom("size") it may block, but the second one readFrom("object", 0) doesn't, because both are protected by the same TestUntil created with afterCompleting(1).
To create properties, register both writers and readers. The writers are all java.util.function.Consumer<T> expressions. The readers may be simple java.util.function.Supplier<T> expressions or java.util.function.Function<T,R> expressions. The properties are now available on both sides of the test.
Again inside the MockPersistResultInterest create the code that handles writing to properties when it receives a notification of some persistence results. It uses a property writer to set the value atomically and inside a lock. You don't have to know the details of the implementation inside AccessSafely, only that thread-safe property access is handled for you.
public void persistResultedIn(
final Outcome outcome,
final Object persistentObject,
final int possible,
final int actual,
final Object object) {
if (actual == 1) {
access.writeUsing("add", persistentObject);
} else if (actual > 1) {
access.writeUsing("addAll", persistentObject);
} else {
throw new IllegalArgumentException(...);
}
}This ensures that a barrier/fence is placed around each data property so each property can be both written to and read from in a thread-safe manner. Here is a code snippet from the test that reads the properties to assert on its expectations.
public void testThatObjectPersistsQuerys() {
final AccessSafely persistAllAccess = persistInterest.afterCompleting(1);
final Person person = new Person("Tom Jones", 85);
objectStore.persist(person, persistInterest);
final int persistSize = persistAllAccess.readFrom("size");
assertEquals(1, persistSize);
assertEquals(person, persistAllAccess.readFrom("object", 0));
...
}This simple tool will prevent what may appear to be impossible data values as test outcomes.
Routers offer a way to deliver a message to one or more other actors, according to a specified routing strategy. They allow you to decouple the source of a message from its destination and they function as a branching mechanism in the message channel between actors. Routers can be used to control and improve message throughput for a particular message protocol, to distribute workload among a pool of actors in a desirable way, or even both.
Based on it's routing strategy, a router will compute a "routing" which consists of an ordered list of one or more actors, referred to as "routees", to which a message will be dispatched for processing. The router may consider any state, including the message itself, in computing the routing.
Routers have four main responsibilities:
Maintain the list of actors that are currently subscribed as routees
Support the subscribing and unsubscribing of routees
Computing a routing for a given message according to some routing logic
Dispatch messages to the actors who were selected to be members of the routing
Routers and routees are actors and, like all actors in vlingo, are type safe. Each will implement one or more messaging protocols (i.e., Java interfaces). A router and its routees must implement the same messaging protocol, though each may also implement other protocols that are not shared.
The io.vlingo.xoom.actors.Router and its abstract subclasses are generic types whose type parameter <P> is the Java interface representing the message protocol implemented by the router and its routees:
public abstract class Router<P> extends ActorXOOM Actors provides several kinds of routers out-of-the-box that are mainly distinguished by the routing strategy they employ:
io.vlingo.xoom.actors.BroadcastRouter - dispatches every message to every routee
io.vlingo.xoom.actors.ContentBasedRouter - considers the content of the message in choosing which routee to dispatch the message to
io.vlingo.xoom.actors.RandomRouter - dispatches each message to a random one of the routees
io.vlingo.xoom.actors.RoundRobinRouter - dispatches each message to the next routee, in turn
io.vlingo.xoom.actors.SmallestMailboxRouter - dispatches each message to the routee with the least number of messages in its mailbox at the time
Each of these routers is provided as an abstract class that extends io.vlingo.xoom.actors.Router (which extends io.vlingo.xoom.actors.Actor).
It is possible, of course, to implement your own new kind of router by directly extending io.vlingo.xoom.actors.Router.
Routees must be subscribed to a router before they can be dispatched any messages. There are two approaches for subscribing routees:
Automatically during router initialization
Explicitly after router initialization
Automatic routee subscription is accomplished by passing a io.vlingo.xoom.actors.RouterSpecification with a non-zero initial pool size to your router's constructor. For more information , see the Creating a Router: Construction and Routee Pool Initialization section below.
Explicit routee subscription requires the router to expose the subscription (and unsubscription) protocol as a public method(s). Since all the routees must comply with the messaging protocol, create an interface providing the type safe subscription protocol, for example:
public interface InvoiceSubmitterSubscription {
void subscribe(InvoiceSubmitter submitter);
void unsubscribe(InvoiceSubmitter submitter);
} The router then needs to implement this interface, self-delegating to the protected subscribe(Routee) and unsubscribe(Routee) methods defined in the io.vlingo.xoom.actors.Router superclass:
public class InvoiceSubmissionRouter extends ..... impements InvoiceSubmitterSubscription, ...{
public void subscribe(InvoiceSubmitter submitter) {
subscribe(Routee.of(submitter));
}
public void unsubscribe(InvoiceSubmitter submitter) {
unsubscribe(Routee.of(submitter));
}
}The io.vlingo.xoom.actors.Router class provides a set of route computation methods:
<T1> Routing<P> routingFor(final T1 routable1)
<T1, T2> Routing<P> routingFor(final T1 routable1, final T2 routable2)
<T1, T2, T3> Routing<P> routingFor(final T1 routable1, final T2 routable2, final T3 routable3)
<T1, T2, T3, T4> Routing<P> routingFor(final T1 routable1, final T2 routable2, final T3 routable3, final T4 routable4)Most types of routers do not base routing decisions on the message being routed because their routing logic is more mechanical (e.g., round robin) or is based on the state of the routees (e.g., smallest mailbox).
For this reason, the default implementation of the above four route computation methods simply self-delegates to the method:
protected abstract Routing<P> computeRouting();You will find that each subclass of io.vlingo.xoom.actors.Router will provide a concrete implementation of this method that encodes its particular routing strategy logic.
The exception is io.vlingo.xoom.actors.ContentBasedRouter for which, by definition, the routing logic depends on the specific message being routed. For this reason, ContentBasedRouter implements computeRouting() to throw java.lang.UnsupportedOperationException. Instead, you will need to override one of the routingFor() methods listed above and encode your routing logic there.
As noted previously, one of the responsibilities of routers is to dispatch messages to the actors that were selected as routees. The abstract class io.vlingo.xoom.actors.Router provides two sets of message dispatching protocol:
Command dispatching
Query dispatching
The command dispatching protocol is used to dispatch messages that may have arguments but do not have a return value, for example void submitInvoice(Invoice):
<T1> void dispatchCommand(final BiConsumer<P, T1> action, final T1 routable1)
<T1, T2> void dispatchCommand(final TriConsumer<P, T1, T2> action, final T1 routable1, final T2 routable2)
<T1, T2, T3> void dispatchCommand(final QuadConsumer<P, T1, T2, T3> action, final T1 routable1, final T2 routable2, final T3 routable3)
<T1, T2, T3, T4> void dispatchCommand(final PentaConsumer<P, T1, T2, T3, T4> action, final T1 routable1, final T2 routable2, final T3 routable3, final T4 routable4)The command dispatching methods take a Consumer (function) as the first argument, and then from 1 to 4 objects that will be the arguments to the Consumer function. These objects map to the arguments of the messaging protocol method.
The query dispatching protocol is used to dispatch messages that may have arguments and do return a value, for example Price computePrice(ProductID, CustomerID):
<T1, R extends Completes<?>> R dispatchQuery(final BiFunction<P, T1, R> query, final T1 routable1)
<T1, T2, R extends Completes<?>> R dispatchQuery(final TriFunction<P, T1, T2, R> query, final T1 routable1, final T2 routable2)
<T1, T2, T3, R extends Completes<?>> R dispatchQuery(final QuadFunction<P, T1, T2, T3, R> query, final T1 routable1, final T2 routable2, final T3 routable3)
<T1, T2, T3, T4, R extends Completes<?>> R dispatchQuery(final PentaFunction<P, T1, T2, T3, T4, R> query, final T1 routable1, final T2 routable2, final T3 routable3, final T4 routable4)The query dispatching methods take a Function as the first argument, and then from 1 to 4 objects that will be the arguments to the Function. These objects map to the arguments of the messaging protocol method.
Implementing a router in your application is straight forward. You'll need to:
Create the message protocol that your router and routees will support as a Java interface
Create your routee actor
Create your router actor by extending one of the abstract subclasses of io.vlingo.xoom.actors.Router
Create the Message Protocol
As an example, let's say you are implementing an invoice submission process and there are several ERP systems to which any given invoice might need to be submitted. Your router will route each Invoice to one of several InvoiceSubmitter actors, each of which represents a particular ERP system. The selected routing will likely depend on information in the Invoice itself (e.g., customer ID).
To support this scenario, you might design the following protocol:
public interface InvoiceSubmitter {
void submitInvoice(Invoice invoice);
}Create the Routee Actor(s)
Now that you have the InvoiceSubmitter protocol defined, you can create the actor(s) to which your router will route messages. These actors will be the ones actually doing the work, in our case submitting an Invoice to a specific ERP system.
Let's assume ERP systems in your company are named after Greek mythological gods, such as Apollo and Atlas, and that for various reasons you decide to create a different actor per ERP system.
Here is the rough definition of the actor for the Apollo ERP invoice submission:
public class ApolloInvoiceSubmitter extends Actor implements InvoiceSubmitter {
public void submitInvoice(Invoice invoice) {
//do some work here
}
}Your actor for Atlas and other ERPs would have a similar structure but different logic in their submitInvoice methods.
Create a Router: Extending a Built-in Router
As noted above, XOOM Actors provides a number of built-in implementations of io.vlingo.xoom.actors.Router that you can extend.
In the example, the routing will be content-based, because the router will use information from the message itself (e.g., the customer ID of the Invoice) to select which ERP-specific routee to dispatch the message to. Thus, t will extend io.vlingo.xoom.actors.ContentBasedRouter:
public class InvoiceSubmissionRouter
extends ContentBasedRouter<InvoiceSubmitter>
implements InvoiceSubmitter {
...
}Creating a Router: Construction and Routee Pool Initialization
The io.vlingo.xoom.actors.Router class requires that a io.vlingo.xoom.actors.RouterSpecification be provided as a constructor argument and offers subclasses an opportunity to get involved in the initialization of the routee pool:
public Router(final RouterSpecification<P> specification) {
this.routees = new ArrayList<Routee<P>>();
initRoutees(specification);
}By default, the information in the RouterSpecification will be used to create an initial (possibly empty) pool of routee actors as child actors of the router actor. Subclasses may customize that behavior by overriding initRoutees(RouterSpecification).
Often, your concrete router class can make these decisions for itself by implementing a constructor that supplies them. For example:
public OrderRouterActor(final int poolSize) {
super(
new RouterSpecification(
poolSize,
Definition.has(OrderRouterWorker.class, Definition.NoParameters()),
OrderRouter.class)
);
}Creating a Router: Routing Logic
In the Route Computation section above, we learned that subclasses of io.vlingo.xoom.actors.ContentBasedRouter need to override one of the routingFor() methods inherited from io.vlingo.xoom.actors.Router and implement the routing logic there. The routingFor() method variants provide access to the message arguments so that the router can inspect that information as part of its routing logic.
In our example, the message protocol void submitInvoice(Invoice invoice) has a single argument, so we can override the single-argument routingFor() method:
@override
protected <T1> Routing<P> routingFor(final T1 routable1) {
Invoice invoice = (Invoice) routable1;
InvoiceSubmitter submitter = selectERPInvoiceSubmitter(invoice.customerID);
return Routing.with(submitter);
}
private InvoiceSubmitter selectERPInvoiceSubmitter(CustomerID customerID) {
//select one of the subscribed InvoiceSubmitter actors based on CustomerID
}Note that both InvoiceSubmissionRouter and ERPInvoiceSubmitter implement the InvoiceSubmitter interface, thereby establishing the type safe messaging protocol that they share, which allows the router to forward messages to the routees in a type safe way.
Creating a Router: Dispatching
Implementing dispatching logic requires implementing the message protocol and calling one of the methods discussed in the Router Message Dispatching section above.
The InvoiceSubmitter protocol declares the one-argument command method void submitInvoice(Invoice), so we we need to provide an implementation that calls the dispatchCommand method inherited from io.vlingo.xoom.actors.Router that accommodates one argument:
protected <T1> void dispatchCommand(final BiConsumer<P, T1> action, final T1 routable1)Notice that the first argument to each dispatchCommand (and dispatchQuery) method is a function. XOOM Actors supplies variants of these dispatching methods for 1 to 4 arity functions. Any lambda expression matching the signature of the dispatching method can be passed.
Here, then, is the dispatching logic of our InvoiceSubmissionRouter:
@Override
public void submitInvoice(final Invoice invoice) {
dispatchCommand(InvoiceSubmitter::submitInvoice, invoice);
}In the implementation above, we take advantage of the fact that Java method references are shorthand for lambda expressions that call the same method - in this case the void submitInvoice(Invoice) method of InvoiceSubmitter.
Creating a Router: Subscribing Routees
As noted in the "Creating a Router: Construction and Routee Pool Initialization" section above, one easy way of subscribing routees is to pass a RouterSpecification with a non-zero initial pool size to your router's constructor. The actor Definition in the RouterSpecification will be used by your router to create and subscribe a number of child actors that implement your routing message protocol. This approach works well when the routees can all be instances of the same class.
Sometimes, such as in our invoice submission example, the routees need to be instances of different classes that all implement the same message protocol. In our example, the InvoiceSubmitter actors for the Apollo and Atlas ERPs were implemented as different classes. In such cases, actors need to be explicitly subscribed to the router. This requires the router to provide public un/subscription protocol, as described in the Routee Subscription section above.
