Removed SuspendUntilFirstOfType reactive operator

Author: stidsborg

Core/Cleipnir.ResilientFunctions.Tests/ReactiveTests/SuspensionTests.cs

@@ -1,7 +1,7 @@
+using System;
 using System.Threading.Tasks;
 using Cleipnir.ResilientFunctions.Domain;
 using Cleipnir.ResilientFunctions.Domain.Exceptions;
-using Cleipnir.ResilientFunctions.Reactive;
 using Cleipnir.ResilientFunctions.Reactive.Extensions;
 using Cleipnir.ResilientFunctions.Reactive.Origin;
 using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -20,7 +20,7 @@ public async Task SuspensionExceptionIsThrownWhenNoEventHasBeenEmittedFromLeafOp
             source.SignalNext("world", new InterruptCount(2));
 
             await Should.ThrowAsync<SuspendInvocationException>(
-                () => source.SuspendUntilFirstOfType<int>()
+                () => source.FirstOfType<int>(TimeSpan.Zero)
             );
         }
 
@@ -32,7 +32,7 @@ public async Task EventIsEmittedInResultWhenEventHasBeenEmittedFromLeafOperator(
             source.SignalNext(1, new InterruptCount(2));
             source.SignalNext("world", new InterruptCount(3));
 
-            var next = await source.SuspendUntilFirstOfType<int>();
+            var next = await source.FirstOfType<int>(TimeSpan.Zero);
 
             next.ShouldBe(1);
         }

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/RFunctionTests/MessagingTests.cs

@@ -62,7 +62,7 @@ public async Task FunctionIsSuspendedWhenAwaitedMessageDoesNotAlreadyExist(Task<
             inner: async Task<string> (string _, Workflow workflow) =>
             {
                 var messages = workflow.Messages;
-                return await messages.SuspendUntilFirstOfType<string>();
+                return await messages.FirstOfType<string>(maxWait: TimeSpan.Zero);
             }
         );
 
@@ -148,7 +148,7 @@ public async Task ScheduleInvocationWithPublishResultToSpecifiedFunctionId(Task<
             inner: async Task<string> (string _, Workflow workflow) =>
             {
                 await child.Schedule(childFunctionId.InstanceId.Value, param: "stuff");
-                return await workflow.Messages.SuspendUntilFirstOfType<string>();
+                return await workflow.Messages.FirstOfType<string>(TimeSpan.Zero);
             }
         );
 

Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/RFunctionTests/SuspensionTests.cs

@@ -225,7 +225,7 @@ protected async Task SuspendedFunctionIsAutomaticallyReInvokedWhenEligibleAndWri
             async Task<string> (string param, Workflow workflow) =>
             {
                 var messages = workflow.Messages;
-                var next = await messages.SuspendUntilFirstOfType<string>();
+                var next = await messages.FirstOfType<string>(TimeSpan.Zero);
                 return next;
             }
         );
@@ -271,7 +271,7 @@ protected async Task SuspendedFunctionIsAutomaticallyReInvokedWhenEligibleByWatc
             async Task<string> (string param, Workflow workflow) =>
             {
                 var messages = workflow.Messages;
-                var next = await messages.SuspendUntilFirstOfType<string>();
+                var next = await messages.FirstOfType<string>(TimeSpan.Zero);
                 return next;
             }
         );

Core/Cleipnir.ResilientFunctions/Reactive/Extensions/LeafOperators.cs

@@ -93,24 +93,22 @@ public static Task<Option<T>> SuspendUntilFirstOrNone<T>(this IReactiveChain<T>
                 ? new Option<T>(emits.Single())
                 : Option<T>.NoValue
             );
-    public static Task<T> SuspendUntilFirstOfType<T>(this IReactiveChain<object> s, TimeSpan? maxWait = null)
-        => s.OfType<T>().SuspendUntilFirst(maxWait);
     public static Task<List<T>> SuspendUntilFirsts<T>(this IReactiveChain<T> s, int count, TimeSpan? maxWait = null)
         => s.Take(count).ToList(maxWait: maxWait ?? TimeSpan.Zero);
 
-    public static Task<T> First<T>(this IReactiveChain<T> s)
-        => FirstOrNone(s)
+    public static Task<T> First<T>(this IReactiveChain<T> s, TimeSpan? maxWait = null)
+        => FirstOrNone(s, maxWait)
             .SelectAsync(o => o.HasValue ? o.Value : throw new NoResultException());
     public static Task<T?> FirstOrDefault<T>(this IReactiveChain<T> s)
         => FirstOrNone(s)
             .SelectAsync(o => o.HasValue ? o.Value : default);
-    public static Task<Option<T>> FirstOrNone<T>(this IReactiveChain<T> s)
-        => Firsts(s, count: 1)
+    public static Task<Option<T>> FirstOrNone<T>(this IReactiveChain<T> s, TimeSpan? maxWait = null)
+        => Firsts(s, count: 1, maxWait)
             .SelectAsync(
                 l => l.Any() ? new Option<T>(l.First()) : Option<T>.NoValue
             );
-    public static Task<T> FirstOfType<T>(this IReactiveChain<object> s)
-        => s.OfType<T>().First();
+    public static Task<T> FirstOfType<T>(this IReactiveChain<object> s, TimeSpan? maxWait = null)
+        => s.OfType<T>().First(maxWait);
 
     public static Task<Option<T>> FirstOfType<T>(this IReactiveChain<object> s, string timeoutId, DateTime expiresAt)
         => s.OfType<T>().TakeUntilTimeout(timeoutId, expiresAt).FirstOrNone();
@@ -119,8 +117,8 @@ public static Task<Option<T>> FirstOfType<T>(this IReactiveChain<object> s, stri
 
     public static Task<T> FirstOf<T>(this IReactiveChain<object> s) => s.FirstOfType<T>();
 
-    public static Task<List<T>> Firsts<T>(this IReactiveChain<T> s, int count)
-        => s.Take(count).ToList();
+    public static Task<List<T>> Firsts<T>(this IReactiveChain<T> s, int count, TimeSpan? maxWait = null)
+        => s.Take(count).ToList(maxWait);
 
     #endregion