With akka typed how can an actor sent message to itself periodically?
This topic is based on a conversation between an anonymous user of Akka and our “Ask AI about Akka” bot in the documentation.
In Akka Typed, an actor can send messages to itself periodically using the withTimers function. This function provides a TimerScheduler which can be used to start a timer that sends a message at fixed intervals.
Here is an example of how to use it:
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.Actor;
import scala.concurrent.duration.FiniteDuration;
import java.util.concurrent.TimeUnit;
public abstract class TestActor1 {
private static final String PERIODICAL_KEY = “PERIODICAL_KEY”;
static final class Message {}
public static Behavior<Object> run() {
return Actor.withTimers(timer -> {
timer.startPeriodicTimer(PERIODICAL_KEY, new Message(), new FiniteDuration(5, TimeUnit.SECONDS));
return <name>)
.onMessage(Object.class, (ctx, msg) ->
TestActor2.run()
)<name>();
});
}
public static void main(String[] args) {
Adapter.spawn(ActorSystem.create("test", ConfigFactory.empty()), run(), "testActor");
}
}
In this example, the TestActor1 actor sends a Message to itself every 5 seconds. The startPeriodicTimer function is used to start the timer. The first argument is a key for the timer, the second argument is the message to be sent, and the third argument is the interval between messages.
For more details, you can refer to the Akka Documentation.
Can I start sending periodical message dynamically? For example, when the actor receives a message to start scheduled message, it then starts sending message to itself periodically. Once it receives the message to stop scheduled message, it stops sending periodic messaging.
Yes, you can dynamically start and stop sending periodic messages in Akka Typed. You can use the startTimerWithFixedDelay or startTimerAtFixedRate methods to start sending messages periodically, and the cancel method to stop the timer.
Here is an example in Scala:
import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl.Behaviors
import scala.concurrent.duration._
object MyActor {
sealed trait Command
case object Start extends Command
case object Stop extends Command
case object Tick extends Command
def apply(): Behavior[Command] =
Behaviors.withTimers { timers =>
Behaviors.receiveMessage {
case Start =>
timers.startTimerWithFixedDelay(Tick, 1.second)
Behaviors.same
case Stop =>
timers.cancel(Tick)
Behaviors.same
case Tick =>
// Do something
Behaviors.same
}
}
}
In this example, when the actor receives a Start message, it starts sending Tick messages to itself every second. When it receives a Stop message, it stops the timer.
Can you provide the previous example in java?
Sure, here is the equivalent Java code:
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.TimerScheduler;
import java.time.Duration;
public class MyActor {
interface Command {}
public enum Start implements Command {
INSTANCE
}
public enum Stop implements Command {
INSTANCE
}
public enum Tick implements Command {
INSTANCE
}
public static Behavior<Command> create() {
return Behaviors.withTimers(timers -> new MyActor(timers).idle());
}
private final TimerScheduler<Command> timers;
private MyActor(TimerScheduler<Command> timers) {
this.timers = timers;
}
private Behavior<Command> idle() {
return Behaviors.receive(Command.class)
.onMessage(Start.class, message -> onStarted())
.onMessage(Stop.class, message -> onStopped())
.onMessage(Tick.class, message -> onTick())
.build();
}
private Behavior<Command> onStarted() {
timers.startTimerWithFixedDelay(Tick.INSTANCE, Duration.ofSeconds(1));
return Behaviors.same();
}
private Behavior<Command> onStopped() {
timers.cancel(Tick.INSTANCE);
return Behaviors.same();
}
private Behavior<Command> onTick() {
// Do something
return Behaviors.same();
}
}
In this example, when the actor receives a Start.INSTANCE message, it starts sending Tick.INSTANCE messages to itself every second. When it receives a Stop.INSTANCE message, it stops the timer.
Can I start sending periodical message dynamically? For example, when the actor receives a message to start scheduled message, it then starts sending message to itself periodically. Once it receives the message to stop scheduled message, it stops sending periodic messaging.
Sure, here is an example of how you can implement this in Java using Akka Typed:
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.TimerScheduler;
import java.time.Duration;
import java.util.Random;
public class MyActor {
interface Command {}
public enum StartValidation implements Command {
INSTANCE
}
public enum Validate implements Command {
INSTANCE
}
public enum StopValidation implements Command {
INSTANCE
}
public static Behavior<Command> create() {
return Behaviors.withTimers(timers -> new MyActor(timers).idle());
}
private final TimerScheduler<Command> timers;
private MyActor(TimerScheduler<Command> timers) {
this.timers = timers;
}
private Behavior<Command> idle() {
return Behaviors.receive(Command.class)
.onMessage(StartValidation.class, message -> onStartValidation())
.onMessage(Validate.class, message -> onValidate())
.onMessage(StopValidation.class, message -> onStopValidation())
.build();
}
private Behavior<Command> onStartValidation() {
timers.startTimerWithFixedDelay(Validate.INSTANCE, Duration.ofSeconds(5));
return Behaviors.same();
}
private Behavior<Command> onValidate() {
Random rand = new Random();
int num = rand.nextInt();
if (num % 2 != 0) {
timers.startSingleTimer(StopValidation.INSTANCE, Duration.ZERO);
}
return Behaviors.same();
}
private Behavior<Command> onStopValidation() {
timers.cancel(Validate.INSTANCE);
return Behaviors.same();
}
}
In this example, when the actor receives a StartValidation.INSTANCE message, it starts sending Validate.INSTANCE messages to itself every 5 seconds. When it receives a Validate.INSTANCE message, it generates a random integer and checks if it’s odd. If the integer is odd, it sends a StopValidation.INSTANCE message to itself. When it receives a StopValidation.INSTANCE message, it stops the timer.
Noticed that this actor does not extend AbstractBehavior, what’s the reason?
The actor in the provided example does not extend AbstractBehavior because it’s not necessary for this specific use case. The AbstractBehavior class is a base class that can be extended to implement an actor using the object-oriented style. It provides a createReceive method that you can override to define the actor’s behavior.
However, in the provided example, the actor’s behavior is defined using the functional style with the Behaviors.receive method. This method returns a Receive behavior that defines how the actor should respond to messages. This style can be more concise and expressive for simple actors, and it doesn’t require extending any base class.
Here’s an example of how you might implement the same actor using AbstractBehavior:
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Receive;
public class MyActor extends AbstractBehavior<MyActor.Command> {
interface Command {}
public enum StartValidation implements Command {
INSTANCE
}
public enum Validate implements Command {
INSTANCE
}
public enum StopValidation implements Command {
INSTANCE
}
public static Behavior<Command> create() {
return Behaviors.setup(MyActor::new);
}
private MyActor(ActorContext<Command> context) {
super(context);
}
@Override
public Receive<Command> createReceive() {
return newReceiveBuilder()
.onMessage(StartValidation.class, this::onStartValidation)
.onMessage(Validate.class, this::onValidate)
.onMessage(StopValidation.class, this::onStopValidation)
.build();
}
private Behavior<Command> onStartValidation() {
// Implementation...
return this;
}
private Behavior<Command> onValidate() {
// Implementation...
return this;
}
private Behavior<Command> onStopValidation() {
// Implementation...
return this;
}
}
In this version, MyActor extends AbstractBehavior and overrides the createReceive method to define its behavior. The onStartValidation, onValidate, and onStopValidation methods are used to handle the different types of messages.
For more details, you can refer to the Akka Documentation.
Please provide the exact implementation in object-oriented way. Pasting the requirements for your reference: Write an actor in java using akka typed, it receives 3 types of messages implementing interface Command. The message types are: StartValidation, Validate, and StopValidation. When receiving StartValidation, the actor starts to send Validate message to itself every 5 seconds. When receive validate, the actor gets a random integer and check if the integer is odd. If the integer is odd, it sends StopValidation to itself, otherwise it does nothing. When receiving StopValidation message, it stops sending Validate message every 5 seconds.
Sure, here is an example of how you can implement this in Java using Akka Typed with the object-oriented style:
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.actor.typed.javadsl.TimerScheduler;
import java.time.Duration;
import java.util.Random;
public class MyActor extends AbstractBehavior<MyActor.Command> {
interface Command {}
public enum StartValidation implements Command {
INSTANCE
}
public enum Validate implements Command {
INSTANCE
}
public enum StopValidation implements Command {
INSTANCE
}
private final TimerScheduler<Command> timers;
public static Behavior<Command> create() {
return Behaviors.setup(context ->
Behaviors.withTimers(timers -> new MyActor(context, timers)));
}
private MyActor(ActorContext<Command> context, TimerScheduler<Command> timers) {
super(context);
this.timers = timers;
}
@Override
public Receive<Command> createReceive() {
return newReceiveBuilder()
.onMessage(StartValidation.class, notUsed -> onStartValidation())
.onMessage(Validate.class, notUsed -> onValidate())
.onMessage(StopValidation.class, notUsed -> onStopValidation())
.build();
}
private Behavior<Command> onStartValidation() {
timers.startTimerWithFixedDelay(Validate.INSTANCE, Duration.ofSeconds(5));
return this;
}
private Behavior<Command> onValidate() {
Random rand = new Random();
int num = rand.nextInt();
if (num % 2 != 0) {
timers.startSingleTimer(StopValidation.INSTANCE, Duration.ZERO);
}
return this;
}
private Behavior<Command> onStopValidation() {
timers.cancel(Validate.INSTANCE);
return this;
}
}
In this example, MyActor extends AbstractBehavior and overrides the createReceive method to define its behavior. The onStartValidation, onValidate, and onStopValidation methods are used to handle the different types of messages.
Note that startPeriodicTimer recommended in the first reply is deprecated, the recommended methods to use are startTimerWithFixedDelay or startTimerAtFixedRate