akkadotnet/akka.net

Port of Akka actors for .NET

Arjen Smits

@Danthar

But even then you'd have come up with a communication protocol. And... well..... before you know it, your redefining something like the SOAP envelope :P

Bartosz Sypytkowski

@Horusiath

@Ciantic Akka.Http is a web framework written on top of akka, not a transport layer for the cluster

@object thx, I'll try to take a look at it this evening

ilhadad

@ilhadad

Ok got rid of the ask pattern and instead used a message scheduler to time out. here is the new code and config on the sender side:
CODE:

            //Attempt to get a list of supervisors from the SupervisorRegistry
            SupervisorRegistryGetListRequest request = new SupervisorRegistryGetListRequest(Self);
            _SupervisorRegistry.Tell(request);

            var timeout = Context.System.Scheduler.ScheduleTellOnceCancelable(1000, Self, new SupervisorRegistryGetListEvent(request, null, false), Self);

            Receive<SupervisorRegistryGetListEvent>(e => {

                if(e.Success)
                {
                    _logger.Info("{0} Received supervisor list.", Self.Path.ToStringWithAddress());

                    timeout.Cancel();

                    // Save the list for internal use
                    _AreaToSupervisorActor = e.ResponseGetList.SupervisorDictionary;

                    _logger.Debug("Actor {0} is Initialized, moving to Ready state.", _ActorType);
                    Become(Ready);

                }
                else
                {
                    _logger.Warning("{0} Cannot retrieve list of supervisors. Unable to initialize.  {1} retries.", Self.Path.ToStringWithAddress(), _FetchSupervisorListReties);

                    // retry the request and increase the timeout
                    _FetchSupervisorListReties++;

                    // Set up the timeout
                    timeout = Context.System.Scheduler.ScheduleTellOnceCancelable(1000*_FetchSupervisorListReties, Self, new SupervisorRegistryGetListEvent(request, null, false), Self);

                    // Send the request again
                    _SupervisorRegistry.Tell(request);

                }

and the config on the sender side:

akka {
                    # here we are configuring log levels
                    log-config-on-start = off
                    stdout-loglevel = DEBUG
                    loglevel = DEBUG

          // Define an Nlog logger for the Akka system
          loggers = ["Akka.Logger.NLog.NLogLogger, Akka.Logger.NLog"]

            actor {
              provider = "Akka.Remote.RemoteActorRefProvider, Akka.Remote"
            }

            remote {
                            log-remote-lifecycle-events = DEBUG
                            log-received-messages = on

            helios.tcp {
                                hostname = "127.0.0.1"
                                port = 8777
              }
            }

And here is the new config and code on the receiver side:

        private void HandleGetListRequest(SupervisorRegistryGetListRequest r)
        {
            ImmutableDictionary<MicroServices.Area,IActorRef> immutableDictOfSupervisorsActors = 
                _KnownSupervisorsActors.ToImmutableDictionary(kvp => kvp.Key, kvp => kvp.Value.SupervisorActorReference);

            Sender.Tell(new SupervisorRegistryGetListResponse(r.Requestor,immutableDictOfSupervisorsActors,r));

        }

                akka {
                    # here we are configuring log levels
                    log-config-on-start = off
                    stdout-loglevel = DEBUG
                    loglevel = DEBUG

          // Define an Nlog logger for the Akka system
          loggers = ["Akka.Logger.NLog.NLogLogger, Akka.Logger.NLog"]

          // Enables connectivity to the remote ActorSystemBridge
          actor {
              provider = "Akka.Remote.RemoteActorRefProvider, Akka.Remote"
          }

          remote {
            helios.tcp {
                port = 8888
                hostname = "127.0.0.1"
            }
          }

                }

@Horusiath I took out the cluster

Ok got rid of the ask pattern and instead used a message scheduler to time out. here is the new code and config on the sender side:
CODE

            //Attempt to get a list of supervisors from the SupervisorRegistry
            SupervisorRegistryGetListRequest request = new SupervisorRegistryGetListRequest(Self);
            _SupervisorRegistry.Tell(request);

            var timeout = Context.System.Scheduler.ScheduleTellOnceCancelable(1000, Self, new SupervisorRegistryGetListEvent(request, null, false), Self);

            Receive<SupervisorRegistryGetListEvent>(e => {

                if(e.Success)
                {
                    _logger.Info("{0} Received supervisor list.", Self.Path.ToStringWithAddress());

                    timeout.Cancel();

                    // Save the list for internal use
                    _AreaToSupervisorActor = e.ResponseGetList.SupervisorDictionary;

                    _logger.Debug("Actor {0} is Initialized, moving to Ready state.", _ActorType);
                    Become(Ready);

                }
                else
                {
                    _logger.Warning("{0} Cannot retrieve list of supervisors. Unable to initialize.  {1} retries.", Self.Path.ToStringWithAddress(), _FetchSupervisorListReties);

                    // retry the request and increase the timeout
                    _FetchSupervisorListReties++;

                    // Set up the timeout
                    timeout = Context.System.Scheduler.ScheduleTellOnceCancelable(1000*_FetchSupervisorListReties, Self, new SupervisorRegistryGetListEvent(request, null, false), Self);

                    // Send the request again
                    _SupervisorRegistry.Tell(request);

                }

and the config on the sender side:


akka {
                    # here we are configuring log levels
                    log-config-on-start = off
                    stdout-loglevel = DEBUG
                    loglevel = DEBUG

          // Define an Nlog logger for the Akka system
          loggers = ["Akka.Logger.NLog.NLogLogger, Akka.Logger.NLog"]

            actor {
              provider = "Akka.Remote.RemoteActorRefProvider, Akka.Remote"
            }

            remote {
                            log-remote-lifecycle-events = DEBUG
                            log-received-messages = on

            helios.tcp {
                                hostname = "127.0.0.1"
                                port = 8777
              }
            }

ilhadad

@ilhadad

How can I examine dead letters to see what's going on? Also how do I get the logger to give me more info as to the root issue? Any other suggestions?

Kris Schepers

@schepersk

Any idea when the new SqlServer persistence package will be available?

ilhadad

@ilhadad

@Horusiath I was able to subscribe to the event stream to pull the dead letters it seems the messages are arriving at the remote system but they are all landing in deadletters. Not sure how to figure out what's wrong by looking at the deadletters. Any suggestions?

Bartosz Sypytkowski

@Horusiath

@ilhadad either the message recipient is dead, was not created, or its address is wrong

@schepersk we'll talk about this probably on today's meeting

Kris Schepers

@schepersk

@Horusiath :+1:

ilhadad

@ilhadad

@Horusiath When I send the reply with an immutable dictionary embedded in it fails, however, when I send the reply with a null it works. I though maybe it was a message size issue but that did not work either. Any other ideas?

Bartosz Sypytkowski

@Horusiath

@ilhadad maybe message cannot be serialized/deserialized?

ilhadad

@ilhadad

I guess so. Is there a way to peek into this process?

Bartosz Sypytkowski

@Horusiath

var serializer = actorSystem.Serialization.FindSerializerFor(message);
var binary = serializer.ToBinary(message);

ilhadad

@ilhadad

Is there a way to decorate a property with an attribute that will force serialization?

ilhadad

@ilhadad

@Horusiath Thanks for your help - we figured it out. The issue is related to the serialization/deserialize process! As it turns out if you have a dictionary of some object the object must have a default constructor. If it does not somehow the deserialization process fails. We even tried sending the message without the default constructor and that too also failed. SO... a default constructor is needed. Painful!

Bartosz Sypytkowski

@Horusiath

@ilhadad strange, I've got a lot of messages without default constructors and I'm pretty sure it was never an issue

ilhadad

@ilhadad

@Horusiath It also seems that the immutabledictionary cannot be sent over the wire while a regular dictionary is not a problem. Wierd. Can I use a regular dictionary?

Bartosz Sypytkowski

@Horusiath

yes, unless you modify it after putting into a message. If you have that risk, simply do .ToDictionary() call to create a copy of it.

ilhadad

@ilhadad

ok thanks for all your help.

Matthew Ma

@matthewma7

To handle remote disconnection I did something like this

signalRSelection = Context.ActorSelection("akka.tcp://api@127.0.0.1:4545/user/signalr");
signalRSelection.Tell(message);

So, I am forcing a Context.ActorSelection before tell. I am thinking this should work although there could be performance issue.
However, this doesn't seems to work all the time. (It works the first time the remote disconnected).
Anyway to do this kind of if(disconneted){ connect(); }?
(I am aware the watch for Terminated method. However, I have trouble finding the right time to ActorSelection again, because the remote could be still not available. )

Matthew Ma

@matthewma7

(I tried Cluster too, however, a bug regarding cluster broadcast group is prevent me to use it).

Vagif Abilov

@object

I've read recommendations to be extra careful when choosing serializer for persistent actors, and it makes sense - nobody wants past events to become unreadable. So I am rethinking our persistence serializer choice - we switched to Wire after Akka.NET 1.0.7 came with a warning that JSON.NET will be obsolete, but I am becoming more in favour of plain JSON documents that we will always be able to retrieve and parse into some internal type. So I am a bit confused with these recent Akka warnings about switching to Wire. Should we really? What will be the consequences for using JSON.NET? Or should we just use a different lib to read/write JSON documents?

Bartosz Sypytkowski

@Horusiath

@object general advise here is to have separate serializers for inter-cluster communication and for persistence, as these two things have a different priorities. When it comes to JSON.NET - it's a nightmare for case of Akka.NET. It's very slow, has big memory footprint, and it's unreliable in many cases (especially F# data structures and some weird combinations of system.collections.immutable and surrogate mechanism).

Vagif Abilov

@object

@Horusiath so what would you recommend for persistence serializer to ensure long lifetime of encoded events? And if Json docs are OK what handles them better than JSON.NET?

Bartosz Sypytkowski

@Horusiath

it depends on you. Some people like to use json as format for event sourcing - I personally would choose it only when it has native support from database (i.e. postgres, mongodb, eventstore). Otherwise I would go for something schema-driven, like Protocol Buffers, MS Bond or Avro. But tbh it's a matter of personal opinion.

Vagif Abilov

@object

I see. I've been also looking at Protocol Buffers. And in case of Json ServiceStack seems to be quite fast. When it comes to language type support, I am drifting to a conclusion that we should say goodbye to fancy F# types as journal event types and instead use adapters (like some Scala Akka article suggests) to convert to representation close to serialization protocol.

Kris Schepers

@schepersk

Hmm, if a name my sql server persistence plugin something else then "sql-server", it doesn't seem to work.. Anyone else having this problem?

Kris Schepers

@schepersk

akka {
  persistence {
    journal {
      sql-server {
    class = "Akka.Persistence.SqlServer.Journal.SqlServerJournal, Akka.Persistence.SqlServer"
    plugin-dispatcher = "akka.actor.default-dispatcher"
    connection-string = "Data Source=(local);Initial Catalog=HelloClusterSharding;Integrated Security=SSPI"
    connection-timeout = 30s
    schema-name = Sharding
    table-name = EventJournal
    auto-initialize = on
    timestamp-provider = "Akka.Persistence.Sql.Common.Journal.DefaultTimestampProvider, Akka.Persistence.Sql.Common"
    metadata-table-name = Metadata
      }
    }
  }
}

this works..

but when I replace the "sql-server" with something else, it doesn't..

Kris Schepers

@schepersk

I'm trying to use a different sql server persistence for cluster sharding.
I have my sharding config set up like this:

cluster {
  sharding {
    least-shard-allocation-strategy.rebalance-threshold = 3
    journal-plugin-id = "akka.persistence.journal.sql-server"
    snapshot-plugin-id = "akka.persistence.snapshot-store.sql-server"
  }
}

but I would like to replace the "sql-server" name with, let's say "sql-server-sharding"

Damian Reeves

@DamianReeves

If I'm starting a new green field project with Akka and F#, should I start with Akka.FSharp or should I go right to Akkling?

Matthew Ma

@matthewma7

To handle remote disconnection I did something like this
signalRSelection = Context.ActorSelection("akka.tcp://api@127.0.0.1:4545/user/signalr");
signalRSelection.Tell(message);
So, I am forcing a Context.ActorSelection before tell. I am thinking this should work although there could be performance issue.
However, this doesn't seems to work all the time. (It works the first time the remote disconnected).
Anyway to do this kind of if(disconneted){ connect(); }?
(I am aware the watch for Terminated method. However, I have trouble finding the right time to ActorSelection again, because the remote could be still not available. )

Association to [akka.tcp://api@127.0.0.1:4545] having UID [221449129] is irrecoverably failed. UID is now quarantined and all messages to this UID will be delivered to dead letters. Remote actorsystem must be restarted to recover from this situation.
It seems that the remote server is quarantined, is there a way to prevent the remote server being quarantined?

Bartosz Sypytkowski

@Horusiath

@schepersk currently journals are identified internally by their type - the workaround here is to create custom journal (i.e. inheriting from Akka.Persistence.SqlServer.Journal.SqlServerJournal, Akka.Persistence.SqlServer) and replace it's type as a class name for configuration section used by cluster sharding

Boban Pavloski

@bobanco

@morioma i would use the actor selection to get the remote actor ref once on example on Actor.PreStart() and watch for the termination of the remote actor, when the remote actor dies, you can introduce a behavior like Reconnecting/ Registering and schedule the reconnect within some interval like 5secs on example.

here is some code which comes out of my mind now:

        protected override void PreStart()
        {

            Context.Become(Registering());
            Self.Tell(new LookupForServiceProvider());//LookupForService provider is just a simple message to tell yourself to start the actor selection

        }

private UntypedReceive Registering()
        {
            return message =>
            {
                if (message is RegisteredToServiceProvider)
                {
                          var registered = (RegisteredToServiceProvider) message;
                    Console.WriteLine("Registered to the service provider, Path: {0}",registered.ServiceProviderRef.Path);

                     //do some useful things here, mb switch to a different behavior where you can operate normaly
                      Context.Watch(registered.ServiceProviderRef);
                    Context.Become(Registered(registered.ServiceProviderRef));
                 }
                  else if (message is UnableToRegister)
                {
                    // we are not able to connect now, try reconnect within some time, and also stay into the same behavior.
                    Context.System.Scheduler.ScheduleTellOnce(_lookupTimeout,Self, new LookupForServiceProvider(), Self);
                }
                 else if (message is LookupForServiceProvider)
                {
                    var self = Context.Self;
                    try
                    {
                        Context.ActorSelection(sericeProviderPath)
                            .ResolveOne(_lookupTimeout)
                            .ContinueWith<IServiceLookupResult>(x =>
                            {
                                try
                                {
                                    var serviceProvider = x.Result;
                                    return new RegisteredToServiceProvider(serviceProvider);

                                }
                                catch (AggregateException ex)
                                {
                                    return new UnableToRegister();
                                }
                            }).PipeTo(self);
                    }
                    catch (Exception)
                    {
                        Self.Tell(new UnableToRegister());
                    }
                }
        }
}

private UntypedReceive Registered(IActorRef serviceProvider)
        {
            return message =>
            {
                  if (message is Terminated)
                {
                    var terminated = (Terminated)message;
                    if (terminated.ActorRef.Equals(serviceProvider))
                    {

                        Context.Become(Registering());
                        Self.Tell(new LookupForServiceProvider());
                        Console.WriteLine("Message Service provider is down, auto reconnect in {0} secs.", _lookupTimeout.TotalSeconds);
                    }
               }

sorry for bad formatting, but i wrote the code directly here, hope this will help you

Matthew Ma

@matthewma7

@bobanco Thank you the method and the code. I will try it out :+1:

Kris Schepers

@schepersk

@Horusiath I see.. I thought I could avoid creating another custom plugin..

Kris Schepers

@schepersk

@Horusiath Also, inheriting from Akka.Persistence.SqlServer.Journal.SqlServerJournal, Akka.Persistence.SqlServer won't work because you can't replace the SqlServerJournalEngine where the settings are loaded..

Kris Schepers

@schepersk

Or am I missing something here?

Kris Schepers

@schepersk

@Horusiath Maybe I should also mention I want to use 2 flavours of sql-server persistence, one for the aggregate event stream, and another one for sharding persistence..

Vagif Abilov

@object

@Horusiath I agree with your remarks about my PR (#1900). Should I submit a new PR and close this one or there is another way?

Marc Piechura

@marcpiechura

@object you can change and push to your branch as you did before, this updates the PR automatically

no need for a new PR

Vagif Abilov

@object

@Silv3rcircl3 thanks!

qwoz

@qwoz

I'd like to send messages to actors on a schedule and I'd also like to persist this between system restarts. Has anybody come up with a clever solution to do that, taking into account when things were last run, when they're due to run again, etc.?

I'm aware of schedule tell... this is more about the restartable persistence aspect of it.

qwoz

@qwoz

The scenario if it helps: I'm building a reminder kind of service that anybody can use to remind them to do something and there could end up being tens of thousands of reminders at any time. The reminders might be for one message 20 minutes from now, might be every 5 minutes, might be every day at 8:03 AM, might be every last Friday of every second month at midnight. Accuracy isn't too important so if something is off by some number of seconds or minutes that's okay. But outages, restarts, etc. must not cause a reminder to be missed. Basically some kind of reliable cron daemon for actors.

Where communities thrive

People

Repo info

Activity