dotnet testthe first one crashes, because I haven't implemented deleting messages yet, the second just says "Test run in progress." It seems the process just won't stop.
var props = Context.DI().Props<ProcessUserSessionActor>()
.WithRouter(new ConsistentHashingPool(10000)
.WithHashMapping(b => ((ProcessUserSession)b).UserId.ToString()));
Receive<ProcessUserSession>(a =>
{
var userSessionProcessor = Context.ActorOf(props);
userSessionProcessor.Tell(a);
});
Context.ActorOf when it receives the first message. It seems that if my key is the number 1 then it creates 10,000 instances with hash key 1. I have the nrOfInstances variable set to 1 so I'm going to have to test it to see if the pool isn't capped at 1 routee.
@ChartlogInc_twitter : Maybe not the best analogy but I like to think of Props as like a 'serialized constructor call'. The props don't do anything until they are used in an ActorOf type context. (This is how remote deployment functions; the Props are sent over the wire to the system the actors will run on.)
Without trying to make too many assumptions about what you're trying to do, I'd suggest declaring something like:
private IActorRef _userSessionProcessorPool;
in your class, and then placing _userSessionProcessorPool = Context.ActorOf(props); outside of your receive method (but still inside the constructor.)
New one for today:
private static SupervisorStrategy ProducerSupervisorStrategory = new OneForOneStrategy(
maxNrOfRetries: 5,
withinTimeRange: TimeSpan.FromSeconds(30),
localOnlyDecider: ex => Directive.Stop
); _sqsProducer = Context.ActorOf(Context.DI().Props<SqsProducer>().WithSupervisorStrategy(ProducerSupervisorStrategory), "producer");Supervisor strategy makes no difference when the actor throws an exception. It restarts indefinitely. I even tried swapping out for StoppingSupervisorStrategy and it does the same.
@glikoz Hi Depending how many messages you received and need to handle, akka stream could make your task much easier in particular to maintain time line / order / consistency of the different events you receive (for multiple in instruments I assume).
Akka stream is also amazing at controling the stream when you replay data and want to go as fast as possible as opposed to live where you actually do not control the flow speed (and most likely will be slower than when you replay data yourself)
Based on my experience, operators on the system that I found very usefull to do the transformations are
new AccumulateWhileUnchanged (from alpakka) to group batch of events and aggregate per time or event order say
throttle to control the flow of your batches of events
scan or scanAsync to transform and update (similar to a fold with every step or transfrom being returned which can be run on arbitrary complex structure as you define it yourself)
map / mapAsync and filter obviously to do some one time transformation
When aggregating multiple streams via fan in Zip or ZipLatest with filter is much better than mergeSorted or mergeSortedN which can be tricky as you need to wait on events on all ports
The Graph API is very usefull and I almost never had to create a custom graphStage
Colin breck website was an amazing ressource for understanding stream / examples
public class MainActor : ReceiveActor
{
public MainActor()
{
var sw = Stopwatch.StartNew();
var actorRefList = new List<IActorRef>();
var taskList = new List<Task<int>>();
foreach (var item in Enumerable.Range(0, 50))
{
var actorRef = Context.ActorOf(Props.Create(() => new Actor()), $"{Guid.NewGuid()}");
actorRefList.Add(actorRef);
taskList.Add(actorRef.Ask<int>(1));
}
Receive<int[]>(msg => { sw.Stop(); Console.WriteLine(sw.ElapsedMilliseconds); });
Receive<string>(msg =>
{
sw.Restart();
var xx = Task.WhenAll(taskList.ToArray())
.ContinueWith(_ => {
sw.Stop();
return taskList.Select(p => p.Result).ToArray();
} )
.PipeTo(Self);
});
}
}