Control over recipients lifetime (#16)

samples/NScatterGather.Samples.CompetingTasks/NScatterGather.Samples.CompetingTasks.csproj

@@ -13,7 +13,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Spectre.Console" Version="0.33.0" />
+    <PackageReference Include="Spectre.Console" Version="0.35.0" />
   </ItemGroup>
 
 </Project>

samples/NScatterGather.Samples.CompetingTasks/Program.cs

@@ -112,9 +112,9 @@ static bool IsProductOutOfStock(Evaluation evaluation)
 static RecipientsCollection CollectRecipients()
 {
     var collection = new RecipientsCollection();
-    collection.Add<AlibabaSupplier>("Alibaba");
-    collection.Add<AmazonSupplier>("Amazon");
-    collection.Add<WalmartSupplier>("Walmart");
+    collection.Add<AlibabaSupplier>(name: "Alibaba");
+    collection.Add<AmazonSupplier>(name: "Amazon");
+    collection.Add<WalmartSupplier>(name: "Walmart");
     return collection;
 }
 

samples/NScatterGather.Samples/NScatterGather.Samples.csproj

@@ -13,7 +13,7 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="Spectre.Console" Version="0.33.0" />
+    <PackageReference Include="Spectre.Console" Version="0.35.0" />
   </ItemGroup>
 
 </Project>

samples/NScatterGather.Samples/Samples/2.FilterOnResponse.cs

@@ -11,17 +11,19 @@ public async Task Run()
             var collection = new RecipientsCollection();
             collection.Add<Foo>();
             collection.Add<Bar>();
+            collection.Add<Baz>();
 
             var aggregator = new Aggregator(collection);
 
             // The Send<TRequest> method checks only the input
             // parameter of the methods:
             AggregatedResponse<object?> all = await aggregator.Send(42);
 
-            var allResults = all.AsResultsList(); // 42L, "42"
+            var allResults = all.AsResultsList(); // "42", 42L, 42
             Console.WriteLine($"" +
                 $"{allResults[0]} ({allResults[0]?.GetType().Name}), " +
-                $"{allResults[1]} ({allResults[1]?.GetType().Name})");
+                $"{allResults[1]} ({allResults[1]?.GetType().Name}), " +
+                $"{allResults[2]} ({allResults[2]?.GetType().Name})");
 
             // Instead the Send<TRequest, TResponse> method checks
             // also the return type of the methods, allowing to filter
@@ -41,5 +43,10 @@ class Bar
         {
             public long Longify(int n) => n * 1L;
         }
+
+        class Baz
+        {
+            public int Echo(int n) => n;
+        }
     }
 }

samples/NScatterGather.Samples/Samples/3.InvokeAsyncMethods.cs

@@ -20,7 +20,7 @@ public async Task Run()
             // before aggregating the response.
             var response1 = await aggregator.Send(42);
 
-            var results = response1.AsResultsList(); // "42", 42L, 84L
+            var results = response1.AsResultsList(); // 84L, 42L, "42"
             Console.WriteLine($"" +
                 $"{results[0]} ({results[0]?.GetType().Name}), " +
                 $"{results[1]} ({results[1]?.GetType().Name}), " +
@@ -32,9 +32,9 @@ public async Task Run()
             // or ValueTask<TResponse> get invoked.
 
             var response2 = await aggregator.Send<long>(42);
-            var guidResults = response2.AsResultsList();
-            Console.WriteLine($"{guidResults[0]} ({guidResults[0].GetType().Name})");
-            Console.WriteLine($"{guidResults[0]} ({guidResults[1].GetType().Name})");
+            var longResults = response2.AsResultsList();
+            Console.WriteLine($"{longResults[0]} ({longResults[0].GetType().Name})");
+            Console.WriteLine($"{longResults[1]} ({longResults[1].GetType().Name})");
 
             var response3 = await aggregator.Send<string>(42);
             var stringResults = response3.AsResultsList();

samples/NScatterGather.Samples/Samples/5.Timeout.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Threading;
 using System.Threading.Tasks;
 
 namespace NScatterGather.Samples.Samples
@@ -15,20 +14,20 @@ public async Task Run()
 
             var aggregator = new Aggregator(collection);
 
-            // The consumer can limit the duration of the aggregation
-            // by providing a CancellationToken to the aggregator:
+            // The consumer can limit the duration of the aggregation by
+            // providing a timeout (or CancellationToken) to the aggregator:
             // this will ensure that the response will be ready in
             // the given amount of time and will "discard" incomplete
             // (and also never-ending) invocations.
-            using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100));
-            var response = await aggregator.Send(42, cts.Token);
+            var response = await aggregator.Send(42, TimeSpan.FromMilliseconds(50));
 
             // The recipients that didn't complete in time will be
             // listed in the Incomplete property of the result:
             Console.WriteLine($"Completed {response.Completed.Count}");
             Console.WriteLine(
                 $"Incomplete {response.Incomplete.Count}: " +
-                $"{response.Incomplete[0].RecipientType?.Name}");
+                $"{response.Incomplete[0].RecipientType?.Name}, " +
+                $"{response.Incomplete[1].RecipientType?.Name}");
         }
 
         class Foo
@@ -40,7 +39,7 @@ class Bar
         {
             public async Task<int> Long(int n)
             {
-                await Task.Delay(100);
+                await Task.Delay(1000);
                 return n;
             }
         }

samples/NScatterGather.Samples/Samples/6.DelegateRecipients.cs

@@ -17,7 +17,7 @@ public async Task Run()
             var aggregator = new Aggregator(collection);
 
             var responseOfInt = await aggregator.Send(42);
-            var resultsOfInt = responseOfInt.AsResultsList(); // 84, "42"
+            var resultsOfInt = responseOfInt.AsResultsList(); // "42", 84
             Console.WriteLine($"{resultsOfInt[0]}, {resultsOfInt[1]}");
 
             var onlyStrings = await aggregator.Send<string>(42);

src/NScatterGather/Aggregator.cs

@@ -4,17 +4,20 @@
 using System.Threading;
 using System.Threading.Tasks;
 using NScatterGather.Recipients;
+using NScatterGather.Recipients.Collection.Scope;
+using NScatterGather.Recipients.Run;
 using NScatterGather.Responses;
-using NScatterGather.Run;
 
 namespace NScatterGather
 {
     public class Aggregator
     {
-        private readonly RecipientsCollection _recipients;
+        private readonly IRecipientsScope _scope;
 
-        public Aggregator(RecipientsCollection recipients) =>
-            _recipients = recipients;
+        public Aggregator(RecipientsCollection collection)
+        {
+            _scope = collection.CreateScope();
+        }
 
         public async Task<AggregatedResponse<object?>> Send(
             object request,
@@ -31,23 +34,23 @@ public class Aggregator
             if (request is null)
                 throw new ArgumentNullException(nameof(request));
 
-            var recipients = _recipients.ListRecipientsAccepting(request.GetType());
+            var recipients = _scope.ListRecipientsAccepting(request.GetType());
 
             var invocations = await Invoke(recipients, request, cancellationToken)
                 .ConfigureAwait(false);
 
             return AggregatedResponseFactory.CreateFrom(invocations);
         }
 
-        private async Task<IReadOnlyList<RecipientRunner<object?>>> Invoke(
+        private async Task<IReadOnlyList<RecipientRun<object?>>> Invoke(
             IReadOnlyList<Recipient> recipients,
             object request,
             CancellationToken cancellationToken)
         {
-            var runners = recipients.Select(r => new RecipientRunner<object?>(r)).ToArray();
+            var runners = recipients.Select(recipient => recipient.Accept(request)).ToArray();
 
             var tasks = runners
-                .Select(runner => runner.Run(x => x.Accept(request)))
+                .Select(runner => runner.Start())
                 .ToArray();
 
             var allTasksCompleted = Task.WhenAll(tasks);
@@ -76,23 +79,23 @@ public class Aggregator
             if (request is null)
                 throw new ArgumentNullException(nameof(request));
 
-            var recipients = _recipients.ListRecipientsReplyingWith(request.GetType(), typeof(TResponse));
+            var recipients = _scope.ListRecipientsReplyingWith(request.GetType(), typeof(TResponse));
 
             var runners = await Invoke<TResponse>(recipients, request, cancellationToken)
                 .ConfigureAwait(false);
 
             return AggregatedResponseFactory.CreateFrom(runners);
         }
 
-        private async Task<IReadOnlyList<RecipientRunner<TResponse>>> Invoke<TResponse>(
+        private async Task<IReadOnlyList<RecipientRun<TResponse>>> Invoke<TResponse>(
             IReadOnlyList<Recipient> recipients,
             object request,
             CancellationToken cancellationToken)
         {
-            var runners = recipients.Select(r => new RecipientRunner<TResponse>(r)).ToArray();
+            var runners = recipients.Select(recipient => recipient.ReplyWith<TResponse>(request)).ToArray();
 
             var tasks = runners
-                .Select(runner => runner.Run(x => x.ReplyWith<TResponse>(request)))
+                .Select(runner => runner.Start())
                 .ToArray();
 
             var allTasksCompleted = Task.WhenAll(tasks);

src/NScatterGather/Inspection/ConflictException.cs → src/NScatterGather/Inspection/CollisionException.cs

@@ -2,15 +2,15 @@
 
 namespace NScatterGather
 {
-    public class ConflictException : Exception
+    public class CollisionException : Exception
     {
         public Type RecipientType { get; }
 
         public Type RequestType { get; }
 
         public Type? ResponseType { get; }
 
-        internal ConflictException(
+        internal CollisionException(
             Type recipientType,
             Type requestType,
             Type? responseType = null)

src/NScatterGather/Inspection/TypeInspector.cs

@@ -11,14 +11,16 @@ internal class TypeInspector
         private static readonly BindingFlags DefaultFlags = BindingFlags.Public | BindingFlags.Instance;
         private static readonly MethodAnalyzer _methodAnalyzer = new MethodAnalyzer();
 
+        public Type Type => _type;
+
         private readonly MethodMatchEvaluationCache _evaluationsCache = new MethodMatchEvaluationCache();
 
         private readonly IReadOnlyList<MethodInspection> _methodInspections;
         private readonly Type _type;
 
         public TypeInspector(Type type)
         {
-            _type = type;
+            _type = type ?? throw new ArgumentNullException(nameof(type));
             _methodInspections = InspectMethods(type);
 
             // Local functions.
@@ -68,7 +70,7 @@ static IReadOnlyList<MethodInspection> InspectMethods(Type type)
             _evaluationsCache.TryAdd(requestType, new MethodMatchEvaluation(false, method));
 
             if (matches.Count > 1)
-                throw new ConflictException(_type, requestType);
+                throw new CollisionException(_type, requestType);
 
             return false;
         }
@@ -126,7 +128,7 @@ private IReadOnlyList<MethodInfo> ListMatchingMethods(Type requestType)
             _evaluationsCache.TryAdd(requestType, responseType, new MethodMatchEvaluation(false, method));
 
             if (matches.Count > 1)
-                throw new ConflictException(_type, requestType, responseType);
+                throw new CollisionException(_type, requestType, responseType);
 
             return false;
         }

src/NScatterGather/Inspection/TypeInspectorRegistry.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Concurrent;
-using System.Collections.Generic;
 
 namespace NScatterGather.Inspection
 {
@@ -9,10 +8,9 @@ internal class TypeInspectorRegistry
         private readonly ConcurrentDictionary<Type, TypeInspector> _registry =
             new ConcurrentDictionary<Type, TypeInspector>();
 
-        public TypeInspector Register<T>() =>
-            Register(typeof(T));
+        public TypeInspector For<T>() => For(typeof(T));
 
-        public TypeInspector Register(Type type)
+        public TypeInspector For(Type type)
         {
             if (_registry.TryGetValue(type, out var cached))
                 return cached;
@@ -22,17 +20,6 @@ public TypeInspector Register(Type type)
             return inspector;
         }
 
-        public TypeInspector Of<T>() =>
-            Of(typeof(T));
-
-        public TypeInspector Of(Type t)
-        {
-            _ = _registry.TryGetValue(t, out var inspector);
-            return inspector ??
-                throw new KeyNotFoundException($"No {nameof(TypeInspector)} for type '{t.Name}' was registered.");
-        }
-
-        internal void Clear() =>
-            _registry.Clear();
+        internal void Clear() => _registry.Clear();
     }
 }

src/NScatterGather/Internals/EnumBcl.cs

@@ -0,0 +1,16 @@
+using System;
+
+namespace NScatterGather
+{
+    internal static class EnumBcl
+    {
+        public static bool IsValid<TEnum>(this TEnum value) where TEnum : struct, Enum
+        {
+#if NETSTANDARD2_0 || NETSTANDARD2_1
+            return Enum.IsDefined(typeof(TEnum), value);
+#else
+            return Enum.IsDefined(value);
+#endif
+        }
+    }
+}

src/NScatterGather/NScatterGather.csproj

@@ -4,7 +4,7 @@
     <TargetFrameworks>net5.0;netstandard2.1;netstandard2.0</TargetFrameworks>
     <LangVersion>latest</LangVersion>
     <Nullable>enable</Nullable>
-    <Version>0.3.0</Version>
+    <Version>0.4.0</Version>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <GenerateDocumentationFile>false</GenerateDocumentationFile>
   </PropertyGroup>

src/NScatterGather/Recipients/Collection/CollisionHandler.cs

@@ -0,0 +1,4 @@
+namespace NScatterGather
+{
+    public delegate void CollisionHandler(CollisionException ex);
+}