Hey, I recently used Akka.net on a test project for a job (here's the code if you're interested, but I wouldn't say it's a pristine example of how to use Akka by any means https://github.com/matneyx/FileWatcher ), which I unfortunately didn't get. They were kind enough to provide feedback, and one of the areas they hit was about Akka.net...
Uses Akka, which is made for large-scale distributed app development, in a context that doesnt really make sense. . .
(I cut off the rest of the comment because it was specific to the project, and not really related to this question)
Is their comment accurate? Is Akka.net really only useful for large-scale distributed apps, and I've just been using it wrong, or are they mistaken and Akka can be used for smaller scale things without too much overhead?
Receive handler is of type Action<T>, so yes you can make it async (just like any method returning void), but it won't be tread safe. Basically your actor will start processing a message and then immediatelly will pick the next one without waiting for previous to finish first. Essentially you'll end up with race conditions.
Hi, question about actor discovery for local actor system (no remoting).
Common service actors are located in a single ServicesActor:
/user/ServicesActor/Service1
/user/ServicesActor/Service2
/user/ServicesActor/ServiceN
Some logic per instance in a single SomeManager:
/user/SomeManager/Instance1/SomeOrchestrator/Worker1
/user/SomeManager/Instance1/SomeOrchestrator/Worker2
/user/SomeManager/Instance1/SomeOrchestrator/WorkerN
...
/user/SomeManager/InstanceN/SomeOrchestrator/Worker1..N
I need to send messages from "WorkerX" actors to Service1... ServiceN actors. What is the best way to resolve service actors from worker actors?
One solution is to use "ActorPaths" class recommendation, but in this case I'm ending up with ActorSelection in-game, which is not recommended (https://getakka.net/articles/actors/untyped-actor-api.html) for local actor system, though definitely would simplify the discovery. Second option: pass ServiceActor IActorRef to SomeManager, send a message to ServiceActor something like "RequestAllServiceRefs" which will reply with a DTO that contains all actorrefs for all services, then pass that reply into constructor of "InstanceX" actor which in turn will pass it into constructor of "SomeOrchestrator" actor which will pass it to all child "WorkerX" actors. Quite verbose and complicates the initialization of the SomeManager (can't create child actors in PreStart, because reply is not received yet).
What is the best way to achieve this?
You could have the workeractor communicate with the servicesactor to aquire the IActorRef of the Service actor you want. ActorSelection is also an option. It is not recommended as your default way of doing things. Because its slower then directly working with IActorRefs.
public sealed class ServiceList
{
ctor(Service1..... ServiceN)
public IActorRef Service1 {get;}
public IActorRef Service2 {get;}
public IActorRef ServiceN {get;}
}
--You could have the workeractor communicate with the servicesactor to aquire the IActorRef of the Service actor you want
in that case we are back to the ActorSelection which I was trying to avoid, but at the same time given that my case is not so complicated I started wondering how to properly pass those references (the recommendations of from the aforementioned link). Because it didn't look very straightforward/clean I thought maybe there's something about my design, however, it doesn't seem to be way off from examples I've seen. Eventually, it seems I will have to cook something based on ActorSelection :)
- Assuming we start consuming from the beginning
- I am thinking about fsm actors, reacting on the eof message from kafka (become synced)
- produce the message, and wait for the event to come by, or just assume a still synced state
IStash is settable
private void HandleRunTask( RunTask runTask)
{
_counter++;
Console.WriteLine( "Before RunTask" );
RunTask
(
async () =>
{
_counter++;
Console.WriteLine( $"Counter: {_counter}" );
Console.WriteLine( "Before await" );
await Task.Delay( 5000 ).ConfigureAwait( false );
Console.WriteLine( "After await" );
} );
Console.WriteLine( "After RunTask" );
}
public static void RunTask(Func<Task> asyncAction)
{
var context = ActorCell.Current;
if (context == null)
throw new InvalidOperationException("RunTask must be called from an actor context.");
var dispatcher = context.Dispatcher;
//suspend the mailbox
dispatcher.Suspend(context);
ActorTaskScheduler actorScheduler = context.TaskScheduler;
actorScheduler.CurrentMessage = context.CurrentMessage;
Task<Task>.Factory.StartNew(asyncAction, CancellationToken.None, TaskCreationOptions.None, actorScheduler)
.Unwrap()
.ContinueWith(parent =>
{
Exception exception = GetTaskException(parent);
if (exception == null)
{
dispatcher.Resume(context);
context.CheckReceiveTimeout();
}
else
{
context.Self.AsInstanceOf<IInternalActorRef>().SendSystemMessage(new ActorTaskSchedulerMessage(exception, actorScheduler.CurrentMessage));
}
//clear the current message field of the scheduler
actorScheduler.CurrentMessage = null;
}, actorScheduler);
}public override int MaximumConcurrencyLevel
{
get { return 1; }
}A follow up would be:
public sealed class CHRChildActor : UntypedActor
{
private int? processedUserId;
private static int instanceCounter = 0;
protected override void PreStart()
{
Console.WriteLine( Self.Path );
base.PreStart();
instanceCounter++;
}
protected override void OnReceive( object message )
{
message.Match().With<Work>( HandleWork )
.Default( Unhandled );
}
private void HandleWork(Work work)
{
Console.WriteLine( $"Work received for userId {work.UserId}" );
Console.WriteLine( $"Instance counter: {instanceCounter}" );
if (processedUserId == null)
{
processedUserId = work.UserId;
}
if (processedUserId != work.UserId)
{
throw new Exception( "Being reused!" );
}
}
public sealed class Work : IConsistentHashable
{
public Work(int userId)
{
UserId = userId;
}
public int UserId { get; }
public object ConsistentHashKey => UserId;
}
}basically it throws exception if being reused, now, parent sends 5 messages
for (var i = 0; i < 5; i++)
{
_child.Tell( new CHRChildActor.Work( i ) );
}and the way _child actor is created:
_child = Context.ActorOf( Props.Create( () => new CHRChildActor()).WithRouter( new ConsistentHashingPool( 10 ) ) );