Make actor responses optional by default, added ConsistentHash/RoundR…

…obin reply routers
andresgutierrez · Oct 17, 2023 · e8463dd · e8463dd
1 parent 727a739
commit e8463dd
Show file tree

Hide file tree

Showing 23 changed files with 394 additions and 82 deletions.
diff --git a/Nixie.Tests/Actors/PingPongActor.cs b/Nixie.Tests/Actors/PingPongActor.cs
@@ -8,9 +8,9 @@ public PongActor(IActorContext<PongActor, string, string> _)
 
     }
 
-    public Task<string> Receive(string message)
+    public Task<string?> Receive(string message)
     {        
-        return Task.FromResult(message);
+        return Task.FromResult<string?>(message);
     }
 }
 
@@ -35,7 +35,7 @@ public void IncrMessage()
         receivedMessages++;
     }
 
-    public async Task<string> Receive(string message)
+    public async Task<string?> Receive(string message)
     {
         IncrMessage();
 

diff --git a/Nixie.Tests/Actors/ReplyActor.cs b/Nixie.Tests/Actors/ReplyActor.cs
@@ -26,10 +26,10 @@ public void IncrMessage(string id)
             receivedMessages[id]++;
     }
 
-    public Task<string> Receive(string message)
+    public Task<string?> Receive(string message)
     {
         IncrMessage(message);
 
-        return Task.FromResult(message);
+        return Task.FromResult<string?>(message);
     }
 }
diff --git a/Nixie.Tests/Actors/ReplySlowActor.cs b/Nixie.Tests/Actors/ReplySlowActor.cs
@@ -26,7 +26,7 @@ public void IncrMessage(string id)
             receivedMessages[id]++;
     }
 
-    public async Task<string> Receive(string message)
+    public async Task<string?> Receive(string message)
     {
         IncrMessage(message);
 

diff --git a/Nixie.Tests/Actors/RouteeActor.cs b/Nixie.Tests/Actors/RouteeActor.cs
@@ -22,20 +22,15 @@ public RouterMessage(RouterMessageType type, string data)
 
     public int GetHash()
     {
-        ulong hash = 14695981039346656037;
-
-        ReadOnlySpan<char> spanBytes = Data.AsSpan();
-
-        for (int i = 0; i < spanBytes.Length; i++)
+        return Data switch
         {
-            unchecked
-            {
-                hash ^= spanBytes[i];
-                hash *= 0x100000001b3;
-            }
-        }
-
-        return Math.Abs((int)hash);
+            "aaa" => 0,
+            "bbb" => 1,
+            "ccc" => 2,
+            "ddd" => 3,
+            "eee" => 4,
+            _ => 0
+        };
     }
 }
 

diff --git a/Nixie.Tests/Actors/RouteeReplyActor.cs b/Nixie.Tests/Actors/RouteeReplyActor.cs
@@ -0,0 +1,42 @@
+
+namespace Nixie.Tests.Actors;
+
+public sealed class RouterResponse
+{
+    public string Data { get; }
+
+    public RouterResponse(string data)
+    {
+        Data = data;
+    }
+}
+
+public sealed class RouteeReplyActor : IActor<RouterMessage, RouterResponse>
+{
+    private int receivedMessages;
+
+    private readonly IActorContext<RouteeReplyActor, RouterMessage, RouterResponse> context;
+
+    public RouteeReplyActor(IActorContext<RouteeReplyActor, RouterMessage, RouterResponse> context)
+    {
+        this.context = context;
+    }
+
+    public int GetMessages()
+    {
+        return receivedMessages;
+    }
+
+    public void IncrMessage()
+    {
+        receivedMessages++;
+    }
+
+    public Task<RouterResponse?> Receive(RouterMessage message)
+    {        
+        if (message.Type == RouterMessageType.Route)
+            IncrMessage();
+
+        return Task.FromResult<RouterResponse?>(new RouterResponse(message.Data));
+    }
+}
diff --git a/Nixie.Tests/TestRouters.cs b/Nixie.Tests/TestRouters.cs
@@ -1,6 +1,7 @@
 
 using Nixie.Routers;
 using Nixie.Tests.Actors;
+using System.Diagnostics.Metrics;
 
 namespace Nixie.Tests;
 
@@ -62,6 +63,33 @@ public async void TestCreateRoundRobinRouterSlowSend()
         }
     }
 
+    [Fact]
+    public async void TestCreateRoundRobinRouterReply()
+    {
+        using ActorSystem asx = new();
+
+        IActorRef<RoundRobinActor<RouteeReplyActor, RouterMessage, RouterResponse>, RouterMessage, RouterResponse> router =
+            asx.Spawn<RoundRobinActor<RouteeReplyActor, RouterMessage, RouterResponse>, RouterMessage, RouterResponse>("my-router", 5);
+
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "aaa"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "bbb"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "ccc"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "ddd"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "eee"));
+
+        Assert.IsAssignableFrom<RoundRobinActor<RouteeReplyActor, RouterMessage, RouterResponse>>(router.Runner.Actor);
+
+        RoundRobinActor<RouteeReplyActor, RouterMessage, RouterResponse> routerActor = (RoundRobinActor<RouteeReplyActor, RouterMessage, RouterResponse>)router.Runner.Actor!;
+
+        foreach (IActorRef<RouteeReplyActor, RouterMessage, RouterResponse> routee in routerActor.Instances)
+        {
+            Assert.IsAssignableFrom<RouteeReplyActor>(routee.Runner.Actor);
+
+            RouteeReplyActor routeeActor = (RouteeReplyActor)routee.Runner.Actor!;
+            Assert.Equal(1, routeeActor.GetMessages());
+        }
+    }
+
     [Fact]
     public async void TestCreateConsistentHashRouter()
     {
@@ -70,11 +98,11 @@ public async void TestCreateConsistentHashRouter()
         IActorRef<ConsistentHashActor<RouteeActor, RouterMessage>, RouterMessage> router =
             asx.Spawn<ConsistentHashActor<RouteeActor, RouterMessage>, RouterMessage>("my-router", 5);
 
-        router.Send(new RouterMessage(RouterMessageType.Route, "cf3be82f-7509-4450-9fc6-747490b4696d")); // 2
-        router.Send(new RouterMessage(RouterMessageType.Route, "f9d4c32c-57ea-4826-a632-2bbaf349679e")); // 4
-        router.Send(new RouterMessage(RouterMessageType.Route, "11e27c2a-f8a1-4d8c-a350-7c81b5deb29e")); // 1
-        router.Send(new RouterMessage(RouterMessageType.Route, "35000df2-df97-4a90-96f9-78887965c9e8")); // 0
-        router.Send(new RouterMessage(RouterMessageType.Route, "eac37733-0ec9-440b-98af-553fb65b189f")); // 3
+        router.Send(new RouterMessage(RouterMessageType.Route, "aaa"));
+        router.Send(new RouterMessage(RouterMessageType.Route, "bbb"));
+        router.Send(new RouterMessage(RouterMessageType.Route, "ccc"));
+        router.Send(new RouterMessage(RouterMessageType.Route, "ddd"));
+        router.Send(new RouterMessage(RouterMessageType.Route, "eee"));
 
         await asx.Wait();
 
@@ -90,4 +118,65 @@ public async void TestCreateConsistentHashRouter()
             Assert.Equal(1, routeeActor.GetMessages());
         }
     }
+
+    [Fact]
+    public async void TestCreateConsistentHashRouterReply()
+    {
+        using ActorSystem asx = new();
+
+        IActorRef<ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>, RouterMessage, RouterResponse> router =
+            asx.Spawn<ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>, RouterMessage, RouterResponse>("my-router", 5);
+
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "aaa"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "bbb"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "ccc"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "ddd"));
+        await router.Ask(new RouterMessage(RouterMessageType.Route, "eee"));
+
+        await asx.Wait();
+
+        Assert.IsAssignableFrom<ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>>(router.Runner.Actor);
+
+        ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse> routerActor = (ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>)router.Runner.Actor!;
+
+        foreach (IActorRef<RouteeReplyActor, RouterMessage, RouterResponse> routee in routerActor.Instances)
+        {
+            Assert.IsAssignableFrom<RouteeReplyActor>(routee.Runner.Actor);
+
+            RouteeReplyActor routeeActor = (RouteeReplyActor)routee.Runner.Actor!;
+            Assert.Equal(1, routeeActor.GetMessages());
+        }
+    }   
+
+    [Fact]
+    public async void TestCreateConsistentHashRouterReplyParallel()
+    {
+        using ActorSystem asx = new();
+
+        IActorRef<ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>, RouterMessage, RouterResponse> router =
+            asx.Spawn<ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>, RouterMessage, RouterResponse>("my-router", 5);
+
+        Task[] tasks = new Task[5]
+        {
+            router.Ask(new RouterMessage(RouterMessageType.Route, "aaa")),
+            router.Ask(new RouterMessage(RouterMessageType.Route, "bbb")),
+            router.Ask(new RouterMessage(RouterMessageType.Route, "ccc")),
+            router.Ask(new RouterMessage(RouterMessageType.Route, "ddd")),
+            router.Ask(new RouterMessage(RouterMessageType.Route, "eee")),
+        };
+
+        await Task.WhenAll(tasks);
+
+        Assert.IsAssignableFrom<ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>>(router.Runner.Actor);
+
+        ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse> routerActor = (ConsistentHashActor<RouteeReplyActor, RouterMessage, RouterResponse>)router.Runner.Actor!;
+
+        foreach (IActorRef<RouteeReplyActor, RouterMessage, RouterResponse> routee in routerActor.Instances)
+        {
+            Assert.IsAssignableFrom<RouteeReplyActor>(routee.Runner.Actor);
+
+            RouteeReplyActor routeeActor = (RouteeReplyActor)routee.Runner.Actor!;
+            Assert.Equal(1, routeeActor.GetMessages());
+        }
+    }
 }
diff --git a/Nixie/ActorContextReply.cs b/Nixie/ActorContextReply.cs
@@ -7,7 +7,8 @@ namespace Nixie;
 /// Represents an actor context. This class is passed to the actor when it is created.
 /// It can be used to create other actors or get the sender and the actor system.
 /// </summary>
-public sealed class ActorContext<TActor, TRequest, TResponse> : IActorContext<TActor, TRequest, TResponse> where TActor : IActor<TRequest, TResponse> where TRequest : class where TResponse : class
+public sealed class ActorContext<TActor, TRequest, TResponse> : IActorContext<TActor, TRequest, TResponse> 
+    where TActor : IActor<TRequest, TResponse> where TRequest : class where TResponse : class?
 {
     private readonly ActorSystem actorSystem;
 
@@ -35,6 +36,17 @@ public sealed class ActorContext<TActor, TRequest, TResponse> : IActorContext<TA
     /// </summary>
     public IGenericActorRef? Sender { get; set; }
 
+    /// <summary>
+    /// Returns a reference to the current reply object
+    /// </summary>
+    public ActorMessageReply<TRequest, TResponse>? Reply { get; set; }
+
+    /// <summary>
+    /// Indicates if the response of the current invocation must be by passed
+    /// to allow other consumer to set the response
+    /// </summary>
+    public bool ByPassReply { get; set; }
+
     /// <summary>
     /// Creates a new actor context.
     /// </summary>

diff --git a/Nixie/ActorMessageReply.cs b/Nixie/ActorMessageReply.cs
@@ -8,7 +8,7 @@ namespace Nixie;
 /// <typeparam name="TResponse"></typeparam>
 public sealed record ActorMessageReply<TRequest, TResponse>
 {
-    private int completed = 1;
+    private volatile int completed = 1;
 
     /// <summary>
     /// Returns the request of the message.

diff --git a/Nixie/ActorRefReply.cs b/Nixie/ActorRefReply.cs
@@ -10,7 +10,7 @@ namespace Nixie;
 /// <typeparam name="TRequest"></typeparam>
 /// <typeparam name="TResponse"></typeparam>
 public sealed class ActorRef<TActor, TRequest, TResponse> : IGenericActorRef, IActorRef<TActor, TRequest, TResponse>
-    where TActor : IActor<TRequest, TResponse> where TRequest : class where TResponse : class
+    where TActor : IActor<TRequest, TResponse> where TRequest : class where TResponse : class?
 {
     private readonly ActorRunner<TActor, TRequest, TResponse> runner;
 
@@ -34,7 +34,7 @@ public ActorRef(ActorRunner<TActor, TRequest, TResponse> runner)
     /// <param name="message"></param>
     public void Send(TRequest message)
     {
-        runner.SendAndTryDeliver(message, null);
+        runner.SendAndTryDeliver(message, null, null);
     }
 
     /// <summary>
@@ -44,7 +44,17 @@ public void Send(TRequest message)
     /// <param name="sender"></param>    
     public void Send(TRequest message, IGenericActorRef sender)
     {
-        runner.SendAndTryDeliver(message, sender);
+        runner.SendAndTryDeliver(message, sender, null);
+    }
+
+    /// <summary>
+    /// Passes a message to the actor without expecting a response and specifying a parent promise
+    /// </summary>
+    /// <param name="message"></param>
+    /// <param name="parentPromise"></param>
+    public void Send(TRequest message, ActorMessageReply<TRequest, TResponse>? parentPromise)
+    {
+        runner.SendAndTryDeliver(message, null, parentPromise);
     }
 
     /// <summary>
@@ -54,7 +64,7 @@ public void Send(TRequest message, IGenericActorRef sender)
     /// <returns></returns>
     public async Task<TResponse?> Ask(TRequest message)
     {
-        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, null);
+        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, null, null);
 
         while (!promise.IsCompleted)
             await Task.Yield();
@@ -71,7 +81,7 @@ public void Send(TRequest message, IGenericActorRef sender)
     /// <returns></returns>
     public async Task<TResponse?> Ask(TRequest message, TimeSpan timeout)
     {        
-        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, null);
+        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, null, null);
 
         Stopwatch stopwatch = Stopwatch.StartNew();
 
@@ -94,7 +104,7 @@ public void Send(TRequest message, IGenericActorRef sender)
     /// <returns></returns>
     public async Task<TResponse?> Ask(TRequest message, IGenericActorRef sender)
     {
-        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, sender);
+        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, sender, null);
 
         while (!promise.IsCompleted)
             await Task.Yield();
@@ -112,7 +122,7 @@ public void Send(TRequest message, IGenericActorRef sender)
     /// <returns></returns>
     public async Task<TResponse?> Ask(TRequest message, IGenericActorRef sender, TimeSpan timeout)
     {
-        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, sender);
+        ActorMessageReply<TRequest, TResponse> promise = runner.SendAndTryDeliver(message, sender, null);
 
         Stopwatch stopwatch = Stopwatch.StartNew();
 

diff --git a/Nixie/ActorRepositoryReply.cs b/Nixie/ActorRepositoryReply.cs
@@ -11,7 +11,8 @@ namespace Nixie;
 /// <typeparam name="TActor"></typeparam>
 /// <typeparam name="TRequest"></typeparam>
 /// <typeparam name="TResponse"></typeparam>
-public sealed class ActorRepository<TActor, TRequest, TResponse> : IActorRepositoryRunnable where TActor : IActor<TRequest, TResponse> where TRequest : class where TResponse : class
+public sealed class ActorRepository<TActor, TRequest, TResponse> : IActorRepositoryRunnable 
+    where TActor : IActor<TRequest, TResponse> where TRequest : class where TResponse : class?
 {
     private readonly ActorSystem actorSystem;