remove parallelism-meter form signal-flow and signal-handling-pipelines

falko-team · Jan 15, 2025 · 2d18be7 · 2d18be7
1 parent 2b4b9c9
commit 2d18be7
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 14 deletions.
diff --git a/Sources/Falko.Talkie.Signals/Flows/SignalFlow.cs b/Sources/Falko.Talkie.Signals/Flows/SignalFlow.cs
@@ -14,14 +14,9 @@ public sealed class SignalFlow : ISignalFlow
 
     private readonly RemovableSequence<ISignalHandlingPipeline> _pipelines = new();
 
-    private readonly ParallelismMeter _pipelinesParallelismMeter = new();
-
     private bool _disposed;
 
-    public SignalFlow()
-    {
-        TaskScheduler.UnobservedTaskException += OnUnobservedSignalPublishingException;
-    }
+    public SignalFlow() => TaskScheduler.UnobservedTaskException += OnUnobservedSignalPublishingException;
 
     public Subscription Subscribe(ISignalHandlingPipeline handlingPipeline)
     {
@@ -50,9 +45,10 @@ public async Task PublishAsync(Signal signal, CancellationToken cancellationToke
         try
         {
             // ReSharper disable once InconsistentlySynchronizedField
-            await _pipelines.Parallelize(_pipelinesParallelismMeter)
-                .ForEachAsync((pipeline, scopedCancellationToken) => pipeline.TransferAsync(this, signal, scopedCancellationToken),
-                    cancellationToken: publishCancellationToken);
+            await _pipelines
+                .Parallelize()
+                .ForEachAsync((pipeline, scopedCancellationToken) => pipeline
+                    .TransferAsync(this, signal, scopedCancellationToken), publishCancellationToken);
         }
         catch (SignalPublishingException exception)
         {

diff --git a/Sources/Falko.Talkie.Signals/Pipelines/Handling/ManySignalHandlingPipeline.cs b/Sources/Falko.Talkie.Signals/Pipelines/Handling/ManySignalHandlingPipeline.cs
@@ -15,8 +15,6 @@ public sealed class ManySignalHandlingPipeline
 {
     private readonly FrozenSequence<ISignalHandler> _handlers = handlers.ToFrozenSequence();
 
-    private readonly ParallelismMeter _handlersParallelismMeter = new();
-
     public ValueTask TransferAsync(ISignalFlow flow, Signal signal, CancellationToken cancellationToken = default)
     {
         if (interceptingPipeline?.TryTransfer(signal, out signal, cancellationToken) is false)
@@ -26,9 +24,10 @@ public ValueTask TransferAsync(ISignalFlow flow, Signal signal, CancellationToke
 
         var context = new SignalContext(flow, signal);
 
-        return _handlers.Parallelize(_handlersParallelismMeter)
-            .ForEachAsync((handler, scopedCancellationToken) => handler.HandleAsync(context, scopedCancellationToken),
-                cancellationToken: cancellationToken)
+        return _handlers
+            .Parallelize()
+            .ForEachAsync((handler, scopedCancellationToken) => handler
+                .HandleAsync(context, scopedCancellationToken), cancellationToken: cancellationToken)
             .AsValueTask();
     }
 }