Added an interruptable call chain.

src/Messaging/TagsSubscription.cs

@@ -1,6 +1,6 @@
 using ProSol.Html.Contracts.Data;
 using ProSol.Messaging;
-using ProSol.Messaging.Subsciptions;
+using ProSol.Messaging.Subscriptions;
 
 namespace ProSol.Html.Messaging;
 
@@ -9,13 +9,14 @@ public class TagsSubscription(ISubscriber<TagsProviderMessage> subscriber, param
 {
     protected override void OnNext(ISubscriber<TagsProviderMessage> subscriber, TagsProviderMessage message, NextDelegate next)
     {
+        // HACK: a duplicated code.
         if (tagNames.Any(x => message.CurrentTag.TagInfo.Name == x))
         {
             subscriber.OnNext(message, next);
         }
-
-        // HACK: FilteredSubscription should not break a pipeline 
-        // just because the filter doesn't match.
-        next();
+        else
+        {
+            next();
+        }
     }
 }

src/ProSol.Html.csproj

@@ -8,7 +8,8 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="ProSol.Messaging" Version="3.0.0-rc.0.2" />
+    <PackageReference Include="ProSol.Messaging" Version="3.0.0" />
+    <PackageReference Include="ProSol.Messaging.Subscriptions" Version="1.0.0-rc.5" />
   </ItemGroup>
 
 </Project>

src/TagsProvider.cs

@@ -2,7 +2,7 @@ using ProSol.Html.Contracts.Data;
 using ProSol.Html.Data;
 using ProSol.Html.Messaging;
 using ProSol.Messaging;
-using ProSol.Messaging.Subsciptions;
+using ProSol.Messaging.Subscriptions;
 
 namespace ProSol.Html;
 
@@ -15,7 +15,7 @@ namespace ProSol.Html;
 public class TagsProvider 
 {
     readonly HistoryTracker historyTracker = new();
-    readonly MessageBroker<TagsProviderMessage> broker = new();
+    readonly PipelinePublisher<TagsProviderMessage> publisher = new();
 
     public void Process(ReadOnlySpan<char> html)
     {
@@ -28,15 +28,15 @@ public class TagsProvider
             charsProcessed += Proceed(currentHtml);
         } while (charsProcessed < html.Length);
 
-        broker.Complete();
+        publisher.Complete();
     }    
 
     public IDisposable Subscribe(ISubscriber<TagsProviderMessage> observer)
-        => broker.Subscribe(observer);
+        => publisher.Subscribe(observer);
 
     public IDisposable Subscribe(ISubscriber<TagsProviderMessage> observer, params string[] tagNames)
     {
-        return broker.Subscribe(new TagsSubscription(observer, tagNames));
+        return publisher.Subscribe(new TagsSubscription(observer, tagNames));
     }
 
     void Process(ReadOnlySpan<char> currentHtml, int charsProcessed)
@@ -76,7 +76,7 @@ public class TagsProvider
             [..history], 
             value);
 
-        broker.Publish(message);
+        publisher.Publish(message);
     }
 
     static int Proceed(ReadOnlySpan<char> currentHtml)

src/package.nuspec

@@ -3,7 +3,7 @@
   <metadata>
     <id>ProSol.Html.TagsProvider</id>
     <title>ProSol.Html.TagsProvider</title>
-    <version>2.0.0-rc0.1</version>
+    <version>2.0.0-rc0.3</version>
     <authors>Alex Kozachenko</authors>
     <owners>Alex Kozachenko</owners>
     <requireLicenseAcceptance>false</requireLicenseAcceptance>

tests/TagsProvider_FilteresSubscribers_Tests.cs

@@ -1,9 +1,8 @@
-using ProSol.Html;
 using ProSol.Html.Tests.TestHelpers;
 
 namespace ProSol.Html.Tests;
 
-public class TagsProvider_NamedListeners_Tests
+public class TagsProvider_FilteresSubscribers_Tests
 {
 
     [Test]
@@ -40,9 +39,7 @@ public class TagsProvider_NamedListeners_Tests
         tagsProvider.Subscribe(listener, "i", "b");
 
         var html = """
-        <p>
             <b>Terminal</b> will be reused by <b>tasks</b>, press <i>any</i> key to <i>close</i> it. 
-        </p>
         """;
 
         tagsProvider.Process(html);
@@ -51,6 +48,27 @@ public class TagsProvider_NamedListeners_Tests
         Assert.That(result, Is.EquivalentTo(expected));
     }
 
+    [Test]
+    public void Process_Multiple_NamedSubcribers_ShouldProcess_OnlyOne()
+    {
+        var expected = new string[] { "tasks" };
+
+        var tagsProvider = new TagsProvider();
+        var listener = new TagsProviderListener();
+        var fallbackListener = new TagsProviderListener();
+        tagsProvider.Subscribe(listener, "i", "b");
+        tagsProvider.Subscribe(fallbackListener);
+
+        var html = """
+            <b>Terminal</b> will be reused by <span>tasks</span>, press <i>any</i> key to <i>close</i> it. 
+        """;
+
+        tagsProvider.Process(html);
+
+        var result = GetText(fallbackListener, html);
+        Assert.That(result, Is.EquivalentTo(expected));
+    }
+
     static string[] GetText(TagsProviderListener listener, string html)
     {
         return listener.ProcessedTags

tests/TestHelpers/TagsProviderListener.cs

@@ -27,6 +27,7 @@ internal class TagsProviderListener : ISubscriber<TagsProviderMessage>
     public void OnNext(TagsProviderMessage value, NextDelegate next)
     {
         messages.Add(value);
-        next();
+        // NOTE: this listener gets a message and stops the pipeline, by design.
+        // next();
     }
 }

tests/tests.sln

@@ -0,0 +1,25 @@
+
+Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 17
+VisualStudioVersion = 17.5.002.0
+MinimumVisualStudioVersion = 10.0.40219.1
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ProSol.Html.Tests", "ProSol.Html.Tests.csproj", "{6E961D56-BF63-4177-A8C3-DBD04839CA05}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{6E961D56-BF63-4177-A8C3-DBD04839CA05}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{6E961D56-BF63-4177-A8C3-DBD04839CA05}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{6E961D56-BF63-4177-A8C3-DBD04839CA05}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{6E961D56-BF63-4177-A8C3-DBD04839CA05}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {2DC28F7C-5F68-4E27-B21C-71E206C39F2D}
+	EndGlobalSection
+EndGlobal