Fixes

Author: stidsborg

Cleipnir.Flows.Tests.AspNet/BulkPublishTests.cs

@@ -1,6 +1,5 @@
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Messaging;
-using Cleipnir.ResilientFunctions.Reactive.Extensions;
 using Cleipnir.ResilientFunctions.Storage;
 
 namespace Cleipnir.Flows.Tests.AspNet;
@@ -52,7 +51,7 @@ private class TestFlow : Flow
     {
         public override async Task Run()
         {
-            await Messages.FirstOfType<string>();
+            await Message<string>();
         }
     }
 }

Cleipnir.Flows.Tests/Flows/CorrelationIdFlowTests.cs

@@ -1,5 +1,4 @@
 using Cleipnir.ResilientFunctions.Helpers;
-using Cleipnir.ResilientFunctions.Reactive.Extensions;
 using Cleipnir.ResilientFunctions.Storage;
 using Microsoft.Extensions.DependencyInjection;
 using Shouldly;
@@ -47,7 +46,7 @@ public override async Task Run()
             await Workflow.RegisterCorrelation("SomeCorrelationId");
             CorrelationRegistered = true;
 
-            ReceivedMessage = await Messages.FirstOfType<Message>();
+            ReceivedMessage = await Message<Message>();
         }
     }
 

Cleipnir.Flows.Tests/Flows/FlowsWithResultTests.cs

@@ -1,7 +1,5 @@
-using Cleipnir.ResilientFunctions;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Helpers;
-using Cleipnir.ResilientFunctions.Reactive.Extensions;
 using Cleipnir.ResilientFunctions.Storage;
 using Microsoft.Extensions.DependencyInjection;
 using Shouldly;
@@ -68,28 +66,29 @@ public async Task CompletionOfScheduledFlowCanBeAwaited()
         var flowsContainer = new FlowsContainer(
             flowStore,
             serviceCollection.BuildServiceProvider(),
-            Options.Default
+            new Options(watchdogCheckFrequency: TimeSpan.FromMilliseconds(100))
         );
 
         var flows = new SimpledDelayedFlows(flowsContainer);
         var scheduled = await flows.Schedule("someInstanceId", "someParameter");
 
-        await Task.Delay(1000);
-        var result = await scheduled.Completion();
+        // Flow uses non-suspending delay, so it completes directly
+        var result = await scheduled.Completion(maxWait: TimeSpan.FromSeconds(5));
         result.ShouldBe(1);
     }
-    
+
     private class SimpledDelayedFlows : Flows<SimpledDelayedFlow, string, int>
     {
-        public SimpledDelayedFlows(FlowsContainer flowsContainer) 
+        public SimpledDelayedFlows(FlowsContainer flowsContainer)
             : base(nameof(SimpledDelayedFlow), flowsContainer, options: null) { }
     }
-    
+
     public class SimpledDelayedFlow : Flow<string, int>
     {
         public override async Task<int> Run(string param)
         {
-            await Delay(TimeSpan.FromSeconds(1));
+            // Use non-suspending delay to avoid watchdog dependency
+            await Delay(TimeSpan.FromMilliseconds(100), suspend: false);
             return 1;
         }
     }
@@ -104,7 +103,7 @@ public async Task EventDrivenFlowCompletesSuccessfully()
         var flowsContainer = new FlowsContainer(
             flowStore,
             serviceCollection.BuildServiceProvider(),
-            new Options()
+            new Options(watchdogCheckFrequency: TimeSpan.FromMilliseconds(100))
         );
 
         var flows = new MessageDrivenFuncFlows(flowsContainer);
@@ -113,12 +112,12 @@ public async Task EventDrivenFlowCompletesSuccessfully()
         await Task.Delay(10);
         var controlPanel = await flows.ControlPanel(instanceId: "someInstanceId");
         controlPanel.ShouldNotBeNull();
-        controlPanel.Status.ShouldBe(Status.Executing);
+        controlPanel.Status.ShouldBe(Status.Suspended);
 
         var messageWriter = flows.MessageWriter("someInstanceId");
-        await messageWriter.AppendMessage(2);
+        await messageWriter.AppendMessage(new IntWrapper(2));
 
-        await controlPanel.WaitForCompletion();
+        await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);
 
         await controlPanel.Refresh();
         controlPanel.ShouldNotBeNull();
@@ -136,10 +135,12 @@ public class MessageDrivenFuncFlow : Flow<string, int>
     {
         public override async Task<int> Run(string param)
         {
-            var next = await Messages.FirstOfType<int>(maxWait: TimeSpan.MaxValue);
-            return next;
+            var next = await Message<IntWrapper>();
+            return next.Value;
         }
     }
+
+    public record IntWrapper(int Value);
 
     [TestMethod]
     public async Task FailingFlowCompletesWithError()

Cleipnir.Flows.Tests/Flows/FlowsWithStateTests.cs

@@ -1,7 +1,6 @@
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Helpers;
 using Cleipnir.ResilientFunctions.Messaging;
-using Cleipnir.ResilientFunctions.Reactive.Extensions;
 using Shouldly;
 
 namespace Cleipnir.Flows.Tests.Flows;
@@ -35,11 +34,11 @@ await flows.Run(
             "SomeInstanceId",
             param: "SomeParam",
             new InitialState(
-                [new MessageAndIdempotencyKey("InitialMessageValue")],
-                [new InitialEffect("InitialEffectId", "InitialEffectValue")]
+                [new MessageAndIdempotencyKey(new StringMessage("InitialMessageValue"))],
+                [new InitialEffect(0, "InitialEffectValue")]
             )
         );
-        
+
         flow.InitialEffectValue.ShouldBe("InitialEffectValue");
         flow.InitialMessageValue.ShouldBe("InitialMessageValue");
     }
@@ -48,12 +47,15 @@ private class InitialStateFlow : Flow<string, string>
     {
         public string? InitialEffectValue { get; set; }
         public string? InitialMessageValue { get; set; }
-        
+
         public override async Task<string> Run(string _)
         {
-            InitialEffectValue = await Effect.Get<string>("InitialEffectId");
-            InitialMessageValue = await Messages.OfType<string>().First();
+            InitialEffectValue = await Capture(() => "should not be called");
+            var msg = await Message<StringMessage>();
+            InitialMessageValue = msg.Value;
             return "";
         }
     }
+
+    private record StringMessage(string Value);
 }

Cleipnir.Flows.Tests/Flows/FunctionRegistrationTests.cs

@@ -1,7 +1,6 @@
 using Cleipnir.Flows.AspNet;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Domain.Exceptions;
-using Cleipnir.ResilientFunctions.Reactive.Extensions;
 using Cleipnir.ResilientFunctions.Storage;
 using Microsoft.Extensions.DependencyInjection;
 using Shouldly;
@@ -18,7 +17,10 @@ public async Task SimpleFlowCompletesSuccessfully()
 
         serviceCollection.AddFlows(c => c
             .UseInMemoryStore()
-            .WithOptions(new Options(messagesDefaultMaxWaitForCompletion: TimeSpan.MaxValue))
+            .WithOptions(new Options(
+                messagesDefaultMaxWaitForCompletion: TimeSpan.FromDays(1),
+                watchdogCheckFrequency: TimeSpan.FromMilliseconds(100)
+            ))
             .RegisterFlow<OptionsTestWithOverriddenOptionsFlow, OptionsTestWithOverriddenOptionsFlows>(
                 flowsFactory: sp => new OptionsTestWithOverriddenOptionsFlows(
                     sp.GetRequiredService<FlowsContainer>(),
@@ -34,19 +36,20 @@ public async Task SimpleFlowCompletesSuccessfully()
         await Should.ThrowAsync<InvocationSuspendedException>(
             () => flowsWithOverridenOptions.Run("Id")
         );
-        
+
         var flowsWithDefaultProvidedOptions = sp.GetRequiredService<OptionsTestWithDefaultProvidedOptionsFlows>();
         await flowsWithDefaultProvidedOptions.Schedule("Id");
 
         await Task.Delay(100);
-        
+
         var controlPanel = await flowsWithDefaultProvidedOptions.ControlPanel("Id");
         controlPanel.ShouldNotBeNull();
-        controlPanel.Status.ShouldBe(Status.Executing);
+        // Flow may be Executing (waiting for message) or Suspended depending on timing
+        controlPanel.Status.ShouldBeOneOf(Status.Executing, Status.Suspended);
 
-        await controlPanel.Messages.Append("Hello");
+        await controlPanel.Messages.Append(new StringWrapper("Hello"));
 
-        await controlPanel.WaitForCompletion();        
+        await controlPanel.WaitForCompletion(allowPostponeAndSuspended: true);        
     }
 
 
@@ -60,7 +63,10 @@ public async Task FlowNameCanBeSpecifiedFromTheOutside()
 
         serviceCollection.AddFlows(c => c
             .UseInMemoryStore(store)
-            .WithOptions(new Options(messagesDefaultMaxWaitForCompletion: TimeSpan.MaxValue))
+            .WithOptions(new Options(
+                messagesDefaultMaxWaitForCompletion: TimeSpan.FromDays(1),
+                watchdogCheckFrequency: TimeSpan.FromMilliseconds(100)
+            ))
             .RegisterFlow<SimpleFlow, SimpleFlows>(
                 flowsFactory: sp => new SimpleFlows(
                     sp.GetRequiredService<FlowsContainer>(),
@@ -72,7 +78,7 @@ public async Task FlowNameCanBeSpecifiedFromTheOutside()
         var sp = serviceCollection.BuildServiceProvider();
         var flows = sp.GetRequiredService<SimpleFlows>();
         await flows.Run("Id");
-        var sf = await store.GetFunction(new StoredId(storedType, Instance: "Id".ToStoredInstance()));
+        var sf = await store.GetFunction(StoredId.Create(storedType, "Id"));
         sf.ShouldNotBeNull();
         sf.Status.ShouldBe(Status.Succeeded);
     }
@@ -83,19 +89,21 @@ public class OptionsTestWithOverriddenOptionsFlow : Flow
 {
     public override async Task Run()
     {
-        await Messages.First();
+        await Message<StringWrapper>();
     }
 }
-    
+
 [GenerateFlows]
 public class OptionsTestWithDefaultProvidedOptionsFlow : Flow
 {
     public override async Task Run()
     {
-        await Messages.First();
+        await Message<StringWrapper>();
     }
 }
 
+public record StringWrapper(string Value);
+
 [GenerateFlows]
 public class SimpleFlow : Flow
 {