Making ParamlessRegistration's SendMessage-method more robust

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/Messaging/InMemoryTests/EventSourcesTests.cs

@@ -57,6 +57,10 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task MultipleMessagesCanBeAppendedOneAfterTheOther()
+        => MultipleMessagesCanBeAppendedOneAfterTheOther(FunctionStoreFactory.Create());
+
     [TestMethod]
     public override Task NoOpMessageIsIgnored()
         => NoOpMessageIsIgnored(FunctionStoreFactory.Create());

Core/Cleipnir.ResilientFunctions.Tests/Messaging/TestTemplates/MessagesTests.cs

@@ -469,6 +469,43 @@ await registration.SendMessages(
         await controlPanel1.BusyWaitUntil(c => c.Status == Status.Succeeded);
         await controlPanel2.BusyWaitUntil(c => c.Status == Status.Succeeded);
     }
+    public abstract Task MultipleMessagesCanBeAppendedOneAfterTheOther();
+    protected async Task MultipleMessagesCanBeAppendedOneAfterTheOther(Task<IFunctionStore> functionStoreTask)
+    {
+        var flowType = TestFlowId.Create().Type;
+        var functionStore = await functionStoreTask;
+        using var registry = new FunctionsRegistry(functionStore, new Settings(messagesDefaultMaxWaitForCompletion: TimeSpan.FromSeconds(10)));
+        var messages = new List<string>();
+        var registration = registry.RegisterParamless(
+            flowType,
+            async Task (workflow) =>
+            {
+                await foreach (var message in workflow.Messages)
+                {
+                    if (message is string s)
+                        await workflow.Effect.Capture(() => messages.Add(s));
+                    else
+                        return;
+                }
+            });
+
+        var instanceId = "Instance#1";
+
+        await registration.SendMessage(instanceId, "Hallo");
+        await registration.SendMessage(instanceId, "World");
+        await registration.SendMessage(instanceId, "And");
+        await registration.SendMessage(instanceId, "Universe");
+        await registration.SendMessage(instanceId, -1);
+
+        var cp = await registration.ControlPanel(instanceId).ShouldNotBeNullAsync();
+        await cp.WaitForCompletion(allowPostponeAndSuspended: true);
+        
+        messages.Count.ShouldBe(4);
+        messages[0].ShouldBe("Hallo");
+        messages[1].ShouldBe("World");
+        messages[2].ShouldBe("And");
+        messages[3].ShouldBe("Universe");
+    }
 
     public abstract Task NoOpMessageIsIgnored();
     protected async Task NoOpMessageIsIgnored(Task<IFunctionStore> functionStoreTask)

Core/Cleipnir.ResilientFunctions/ParamlessRegistration.cs

@@ -32,6 +32,7 @@ public class ParamlessRegistration : BaseRegistration
 
     private readonly StateFetcher _stateFetcher;
     public MessageWriters MessageWriters { get; }
+    private readonly IFunctionStore _functionStore;
 
     public ParamlessRegistration(
         FlowType flowType,
@@ -57,6 +58,7 @@ UtcNow utcNow
         _controlPanelFactory = controlPanelFactory;
         MessageWriters = messageWriters;
         _stateFetcher = stateFetcher;
+        _functionStore = functionStore;
     }
 
     public Task<ControlPanel?> ControlPanel(FlowInstance flowInstance)
@@ -79,11 +81,14 @@ public async Task<Finding> SendMessage<T>(
         bool create = true,
         string? idempotencyKey = null) where T : notnull
     {
-        var finding = await Postman.SendMessage(flowInstance.Value.ToStoredInstance(), message, idempotencyKey);
-        if (create && finding == Finding.NotFound)
-            await ScheduleAt(flowInstance, delayUntil: UtcNow());            
+        if (create)
+        {
+            var sf = await _functionStore.GetFunction(new StoredId(StoredType, flowInstance.ToStoredInstance()));
+            if (sf is null)
+                await Schedule(flowInstance);    
+        }
 
-        return finding;
+        return await Postman.SendMessage(flowInstance.Value.ToStoredInstance(), message, idempotencyKey);
     }
 
     public async Task SendMessages(IReadOnlyList<BatchedMessage> messages, bool interrupt = true)

Stores/MariaDB/Cleipnir.ResilientFunctions.MariaDB.Tests/Messaging/MessagesTests.cs

@@ -49,6 +49,10 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task MultipleMessagesCanBeAppendedOneAfterTheOther()
+        => MultipleMessagesCanBeAppendedOneAfterTheOther(FunctionStoreFactory.Create());
+    
     [TestMethod]
     public override Task NoOpMessageIsIgnored()
         => NoOpMessageIsIgnored(FunctionStoreFactory.Create());

Stores/PostgreSQL/Cleipnir.ResilientFunctions.PostgreSQL.Tests/Messaging/MessagesTests.cs

@@ -50,6 +50,10 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task MultipleMessagesCanBeAppendedOneAfterTheOther()
+        => MultipleMessagesCanBeAppendedOneAfterTheOther(FunctionStoreFactory.Create());
+    
     [TestMethod]
     public override Task NoOpMessageIsIgnored()
         => NoOpMessageIsIgnored(FunctionStoreFactory.Create());

Stores/SqlServer/Cleipnir.ResilientFunctions.SqlServer.Tests/Messaging/MessagesTests.cs

@@ -50,6 +50,10 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
 
+    [TestMethod]
+    public override Task MultipleMessagesCanBeAppendedOneAfterTheOther()
+        => MultipleMessagesCanBeAppendedOneAfterTheOther(FunctionStoreFactory.Create());
+    
     [TestMethod]
     public override Task NoOpMessageIsIgnored()
         => NoOpMessageIsIgnored(FunctionStoreFactory.Create());