Refactor Formatters to make Shutdown explicit

CHANGELOG.md

@@ -1,10 +1,11 @@
 # [vNext]
 
 ## Improvements:
-
+* Formatters: Made shutdown an explicit part of the interfaces for the Broker and Formatters. 
+* 
 ## Bug fixes:
 
-*Contributors of this release (in alphabetical order):* 
+*Contributors of this release (in alphabetical order):* @clrudolphi
 
 # v3.3.3 - 2026-01-27
 

Reqnroll/Formatters/FormatterBase.cs

@@ -91,25 +91,23 @@ public async Task PublishAsync(Envelope message)
             return;
         }
 
-        await PostedMessages.Writer.WriteAsync(message);
-
-        // If the _publisher sends the TestRunFinished message, then we can safely shut down.
-        if (message.Content() is TestRunFinished)
+        if (!PostedMessages.Writer.TryWrite(message))
         {
-            _logger.WriteMessage($"DEBUG: Formatters.Plugin {Name} has received the TestRunFinished message and is calling CloseAsync");
-            await CloseAsync();
+            Logger.WriteMessage($"Cannot add message {message.Content().GetType().Name} to formatter {Name} - channel is no longer accepting messages.");
+            return;
         }
     }
 
     protected abstract Task ConsumeAndFormatMessagesBackgroundTask(CancellationToken cancellationToken);
 
-    internal async Task CloseAsync()
+    public async Task CloseAsync()
     {
         Logger.WriteMessage($"DEBUG: Formatters:PluginBase.Close called on formatter {Name}; formatter task was launched: {_formatterTask != null}");
 
         if (PostedMessages.Reader.Completion.IsCompleted || _formatterTask!.IsCompleted)
             throw new InvalidOperationException($"Formatter {Name} has invoked Close when it is already in a closed state.");
 
+        Closed = true;
         PostedMessages.Writer.Complete();
 
         Logger.WriteMessage($"DEBUG: Formatters:PluginBase {Name} has signaled the Channel is closed. Awaiting the writing task.");

Reqnroll/Formatters/ICucumberMessageFormatter.cs

@@ -9,4 +9,5 @@ public interface ICucumberMessageFormatter
     void LaunchFormatter(ICucumberMessageBroker broker);
     string Name { get; }
     Task PublishAsync(Envelope message);
+    Task CloseAsync();
 }

Reqnroll/Formatters/PubSub/CucumberMessageBroker.cs

@@ -104,4 +104,21 @@ public async Task PublishAsync(Envelope message)
         }
     }
 
+    public async Task CompleteAsync()
+    {
+        _logger.WriteMessage("DEBUG: Formatters - Broker: CompleteAsync called. Closing all active formatters.");
+        foreach (var formatter in _activeFormatters.Values)
+        {
+            try
+            {
+                await formatter.CloseAsync();
+            }
+            catch (System.Exception e)
+            {
+                _logger.WriteMessage($"Formatters Broker: Exception closing Formatter Plugin {formatter.Name}: {e.Message}");
+            }
+        }
+        _logger.WriteMessage("DEBUG: Formatters - Broker: CompleteAsync finished.");
+    }
+
 }

Reqnroll/Formatters/PubSub/CucumberMessagePublisher.cs

@@ -194,6 +194,9 @@ internal async Task PublisherTestRunCompleteAsync(TestRunFinishedEvent testRunFi
 
         await _broker.PublishAsync(Envelope.Create(MessageFactory.ToTestRunFinished(TestRunPassed, _clock.GetNowDateAndTime(), _testRunStartedId)));
         _logger.WriteMessage("DEBUG: Formatter:Publisher.TestRunComplete: TestRunFinished Message written");
+
+        await _broker.CompleteAsync();
+        _logger.WriteMessage("DEBUG: Formatter:Publisher.TestRunComplete: Broker CompleteAsync finished");
     }
 
     #region TestThreadExecutionEventPublisher Event Handling Methods

Reqnroll/Formatters/PubSub/ICucumberMessageBroker.cs

@@ -1,9 +1,12 @@
-namespace Reqnroll.Formatters.PubSub;
+using System.Threading.Tasks;
+
+namespace Reqnroll.Formatters.PubSub;
 
 public interface ICucumberMessageBroker : IMessagePublisher
 {
     bool IsEnabled { get; }
 
     void Initialize();
     void FormatterInitialized(ICucumberMessageFormatter formatterSink, bool enabled);
+    Task CompleteAsync();
 }

Tests/Reqnroll.RuntimeTests/Formatters/FormatterBaseTests.cs

@@ -90,14 +90,33 @@ public void LaunchFormatter_Enabled_Calls_LaunchInner_And_StartsTask()
     }
 
     [Fact]
-    public async Task PublishAsync_Writes_Message_And_Closes_On_TestRunFinished()
+    public async Task PublishAsync_Writes_TestRunFinished_Without_Triggering_Shutdown()
     {
         _configMock.Setup(c => c.Enabled).Returns(true);
         _configMock.Setup(c => c.GetFormatterConfigurationByName("testPlugin")).Returns(new Dictionary<string, object>());
         _sut.LaunchFormatter(_brokerMock.Object);
         var msg = Envelope.Create(new TestRunFinished("", false, new Timestamp(0, 0), null, ""));
         await _sut.PublishAsync(msg);
+
+        // TestRunFinished is treated like any other message — no implicit shutdown
         _sut.ConsumedMessages.Should().Contain(msg);
+
+        // Formatter is still open; explicit CloseAsync is required
+        await _sut.CloseAsync();
+    }
+
+    [Fact]
+    public async Task PublishAsync_Logs_When_Channel_No_Longer_Accepting()
+    {
+        _configMock.Setup(c => c.Enabled).Returns(true);
+        _configMock.Setup(c => c.GetFormatterConfigurationByName("testPlugin")).Returns(new Dictionary<string, object>());
+        _sut.LaunchFormatter(_brokerMock.Object);
+        await _sut.CloseAsync();
+
+        // Channel is now completed; PublishAsync should log gracefully via TryWrite failure
+        var msg = Envelope.Create(new TestRunStarted(new Timestamp(0, 0), ""));
+        await _sut.PublishAsync(msg);
+        _loggerMock.Verify(l => l.WriteMessage(It.Is<string>(s => s.Contains("formatter is closed"))), Times.AtLeastOnce);
     }
 
     [Fact]