Added the ability to ignore NoOp messages

Author: stidsborg

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

@@ -56,4 +56,8 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     [TestMethod]
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
+
+    [TestMethod]
+    public override Task NoOpMessageIsIgnored()
+        => NoOpMessageIsIgnored(FunctionStoreFactory.Create());
 }

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

@@ -5,6 +5,7 @@
 using Cleipnir.ResilientFunctions.CoreRuntime;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
 using Cleipnir.ResilientFunctions.Domain;
+using Cleipnir.ResilientFunctions.Domain.Events;
 using Cleipnir.ResilientFunctions.Helpers;
 using Cleipnir.ResilientFunctions.Messaging;
 using Cleipnir.ResilientFunctions.Reactive.Extensions;
@@ -468,6 +469,26 @@ await registration.SendMessages(
         await controlPanel1.BusyWaitUntil(c => c.Status == Status.Succeeded);
         await controlPanel2.BusyWaitUntil(c => c.Status == Status.Succeeded);
     }
+    
+    public abstract Task NoOpMessageIsIgnored();
+    protected async Task NoOpMessageIsIgnored(Task<IFunctionStore> functionStoreTask)
+    {
+        var flowType = TestFlowId.Create().Type;
+        var functionStore = await functionStoreTask;
+        using var registry = new FunctionsRegistry(functionStore);
+        var registration = registry.RegisterFunc<string, string>(
+            flowType,
+            async Task<string> (_, workflow) => (await workflow.Messages.First(maxWait: TimeSpan.FromSeconds(10))).ToString()!
+        );
+
+        var invocation = registration.Invoke("SomeInstance", "SomeParam");
+
+        await registration.SendMessage("SomeInstance", NoOp.Instance);
+        await registration.SendMessage("SomeInstance", "Hallo World!");
+
+        var result = await invocation;
+        result.ShouldBe("Hallo World!");
+    }
 
     private Effect CreateEffect(StoredId storedId, FlowId flowId, IFunctionStore functionStore)
     {

Core/Cleipnir.ResilientFunctions/Domain/Events/NoOp.cs

@@ -0,0 +1,6 @@
+namespace Cleipnir.ResilientFunctions.Domain.Events;
+
+public record NoOp
+{
+    public static NoOp Instance { get; } = new();
+}

Core/Cleipnir.ResilientFunctions/Messaging/MessagesPullerAndEmitter.cs

@@ -6,7 +6,7 @@
 using Cleipnir.ResilientFunctions.CoreRuntime;
 using Cleipnir.ResilientFunctions.CoreRuntime.Invocation;
 using Cleipnir.ResilientFunctions.CoreRuntime.Serialization;
-using Cleipnir.ResilientFunctions.Domain;
+using Cleipnir.ResilientFunctions.Domain.Events;
 using Cleipnir.ResilientFunctions.Reactive.Origin;
 using Cleipnir.ResilientFunctions.Storage;
 
@@ -105,9 +105,11 @@ public async Task PullEvents(TimeSpan maxSinceLastSynced)
 
             storedMessages = filterStoredMessages;
 
-            var events = storedMessages.Select(
-                storedEvent => _serializer.DeserializeMessage(storedEvent.MessageContent, storedEvent.MessageType)
-            );
+            var events = storedMessages
+                .Select(
+                    storedEvent => _serializer.DeserializeMessage(storedEvent.MessageContent, storedEvent.MessageType)
+                )
+                .Where(@event => @event is not NoOp);
 
             Source.SignalNext(events);
         }

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

@@ -48,4 +48,8 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     [TestMethod]
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task NoOpMessageIsIgnored()
+        => NoOpMessageIsIgnored(FunctionStoreFactory.Create());
 }

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

@@ -49,4 +49,8 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     [TestMethod]
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task NoOpMessageIsIgnored()
+        => NoOpMessageIsIgnored(FunctionStoreFactory.Create());
 }

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

@@ -49,4 +49,8 @@ public override Task MessagesRemembersPreviousThrownEventProcessingExceptionOnAl
     [TestMethod]
     public override Task BatchedMessagesIsDeliveredToAwaitingFlows()
         => BatchedMessagesIsDeliveredToAwaitingFlows(FunctionStoreFactory.Create());
+    
+    [TestMethod]
+    public override Task NoOpMessageIsIgnored()
+        => NoOpMessageIsIgnored(FunctionStoreFactory.Create());
 }