Tweaking the UI until it looks cool

src/.editorconfig

@@ -0,0 +1,4 @@
+[*.{csproj,props,targets,xml}]
+indent_style = space
+indent_size = 2
+xml_space_inside_empty_tag = true

src/Billing/Billing.csproj

@@ -14,6 +14,7 @@
 
   <ItemGroup>
     <PackageReference Include="NServiceBus" Version="9.*" />
+    <PackageReference Include="NServiceBus.Persistence.NonDurable" Version="2.*" />
     <PackageReference Include="NServiceBus.Heartbeat" Version="5.*" />
     <PackageReference Include="NServiceBus.Metrics.ServiceControl" Version="5.*" />
   </ItemGroup>

src/Billing/DispatchingProgressBehavior.cs

@@ -0,0 +1,26 @@
+using NServiceBus.Pipeline;
+using NServiceBus.Transport;
+
+namespace Billing;
+
+public class DispatchingProgressBehavior : Behavior<IBatchDispatchContext>
+{
+    private FailureSimulator failureSimulator = new();
+
+    public override async Task Invoke(IBatchDispatchContext context, Func<Task> next)
+    {
+        var incomingMessage = context.Extensions.Get<IncomingMessage>();
+        if (incomingMessage.Headers.ContainsKey("MonitoringDemo.SlowMotion"))
+        {
+            Console.WriteLine($"Dispatching outgoing messages {incomingMessage.MessageId}...");
+            await failureSimulator.RunInteractive(context.CancellationToken);
+        }
+
+        await next().ConfigureAwait(false);
+    }
+
+    public void Failure()
+    {
+        failureSimulator.Trigger();
+    }
+}

src/Billing/FailureSimulation.cs

@@ -0,0 +1,32 @@
+namespace Billing;
+
+public class FailureSimulation
+{
+    private RetrievingMessageProgressBehavior retrievingMessageProgressBehavior = new RetrievingMessageProgressBehavior();
+    private ProcessingMessageProgressBehavior processingMessageProgressBehavior = new ProcessingMessageProgressBehavior();
+    private DispatchingProgressBehavior dispatchingMessageProgressBehavior = new DispatchingProgressBehavior();
+
+    public void Register(EndpointConfiguration endpointConfiguration)
+    {
+        endpointConfiguration.Pipeline.Register(retrievingMessageProgressBehavior, "Shows progress of retrieving messages");
+
+        endpointConfiguration.Pipeline.Register(processingMessageProgressBehavior, "Shows progress of processing messages");
+
+        endpointConfiguration.Pipeline.Register(dispatchingMessageProgressBehavior, "Shows progress of dispatching messages");
+    }
+
+    public void TriggerFailureReceiving()
+    {
+        retrievingMessageProgressBehavior.Failure();
+    }
+
+    public void TriggerFailureProcessing()
+    {
+        processingMessageProgressBehavior.Failure();
+    }
+
+    public void TriggerFailureDispatching()
+    {
+        dispatchingMessageProgressBehavior.Failure();
+    }
+}

src/Billing/FailureSimulator.cs

@@ -0,0 +1,30 @@
+namespace Billing;
+
+public class FailureSimulator
+{
+    private bool failureTriggered = false;
+
+#pragma warning disable PS0003
+    public async Task RunInteractive(CancellationToken cancellationToken)
+#pragma warning restore PS0003
+    {
+        using var progressBar = new ProgressBar();
+
+        for (var i = 0; i <= 100; i++)
+        {
+            if (failureTriggered)
+            {
+                failureTriggered = false;
+                throw new Exception("Simulated failure");
+            }
+            progressBar.Update(i);
+            await Task.Delay(25, cancellationToken).ConfigureAwait(false);
+        }
+    }
+
+    public void Trigger()
+    {
+        //TODO: Use Interlocked
+        failureTriggered = true;
+    }
+}

src/Billing/ProcessingMessageProgressBehavior.cs

@@ -0,0 +1,24 @@
+using NServiceBus.Pipeline;
+
+namespace Billing;
+
+public class ProcessingMessageProgressBehavior : Behavior<IIncomingLogicalMessageContext>
+{
+    private FailureSimulator failureSimulator = new();
+
+    public override async Task Invoke(IIncomingLogicalMessageContext context, Func<Task> next)
+    {
+        if (context.Headers.ContainsKey("MonitoringDemo.SlowMotion"))
+        {
+            Console.WriteLine($"Processing message {context.MessageId}...");
+            await failureSimulator.RunInteractive(context.CancellationToken);
+        }
+
+        await next().ConfigureAwait(false);
+    }
+
+    public void Failure()
+    {
+        failureSimulator.Trigger();
+    }
+}

src/Billing/Program.cs

@@ -1,4 +1,5 @@
-using System.Text.Json;
+using System.Reflection;
+using System.Text.Json;
 using Billing;
 using Messages;
 using Microsoft.Extensions.DependencyInjection;
@@ -16,7 +17,12 @@
         }
 });
 
-endpointConfiguration.UseTransport<LearningTransport>();
+var transport = new LearningTransport
+{
+    StorageDirectory = Path.Combine(Directory.GetParent(Assembly.GetExecutingAssembly().Location)!.Parent!.FullName, ".learningtransport"),
+    TransportTransactionMode = TransportTransactionMode.ReceiveOnly
+};
+endpointConfiguration.UseTransport(transport);
 
 endpointConfiguration.Recoverability()
     .Delayed(delayed => delayed.NumberOfRetries(0));
@@ -34,18 +40,21 @@
     TimeSpan.FromMilliseconds(500)
 );
 
+endpointConfiguration.UsePersistence<NonDurablePersistence>();
+endpointConfiguration.EnableOutbox();
+
+var failureSimulation = new FailureSimulation();
+failureSimulation.Register(endpointConfiguration);
+
 var simulationEffects = new SimulationEffects();
 endpointConfiguration.RegisterComponents(cc => cc.AddSingleton(simulationEffects));
 
 var endpointInstance = await Endpoint.Start(endpointConfiguration);
 
-var nonInteractive = args.Length > 1 && bool.TryParse(args[1], out var isInteractive) && !isInteractive;
-var interactive = !nonInteractive;
-
 UserInterface.RunLoop("Failure rate (Billing)", new Dictionary<char, (string, Action)>
 {
     ['w'] = ("increase the simulated failure rate", () => simulationEffects.IncreaseFailureRate()),
     ['s'] = ("decrease the simulated failure rate", () => simulationEffects.DecreaseFailureRate())
-}, writer => simulationEffects.WriteState(writer), false /* for now*/);
+}, writer => simulationEffects.WriteState(writer));
 
 await endpointInstance.Stop();

src/Billing/ProgressBar.cs

@@ -0,0 +1,21 @@
+namespace Billing;
+
+public class ProgressBar : IDisposable
+{
+    private readonly string widgetId = Guid.NewGuid().ToString("N");
+
+    public ProgressBar()
+    {
+        Console.WriteLine($"!BeginWidget Progress {widgetId}");
+    }
+
+    public void Update(int percent)
+    {
+        Console.WriteLine($"!Widget {widgetId} {percent}");
+    }
+
+    public void Dispose()
+    {
+        Console.WriteLine($"!EndWidget {widgetId}");
+    }
+}

src/Billing/RetrievingMessageProgressBehavior.cs

@@ -0,0 +1,25 @@
+using NServiceBus.Pipeline;
+using Shared;
+
+namespace Billing;
+
+public class RetrievingMessageProgressBehavior : Behavior<ITransportReceiveContext>
+{
+    private FailureSimulator failureSimulator = new();
+
+    public override async Task Invoke(ITransportReceiveContext context, Func<Task> next)
+    {
+        if (context.Message.Headers.ContainsKey("MonitoringDemo.SlowMotion"))
+        {
+            Console.WriteLine($"Retrieving message {context.Message.MessageId}...");
+            await failureSimulator.RunInteractive(context.CancellationToken);
+        }
+
+        await next().ConfigureAwait(false);
+    }
+
+    public void Failure()
+    {
+        failureSimulator.Trigger();
+    }
+}

src/ClientUI/Program.cs

@@ -1,4 +1,5 @@
-using System.Text.Json;
+using System.Reflection;
+using System.Text.Json;
 using ClientUI;
 using Messages;
 using Shared;
@@ -14,7 +15,11 @@
         }
 });
 
-var transport = endpointConfiguration.UseTransport<LearningTransport>();
+var transport = new LearningTransport
+{
+    StorageDirectory = Path.Combine(Directory.GetParent(Assembly.GetExecutingAssembly().Location)!.Parent!.FullName, ".learningtransport")
+};
+var routing = endpointConfiguration.UseTransport(transport);
 
 endpointConfiguration.AuditProcessedMessagesTo("audit");
 endpointConfiguration.SendHeartbeatTo("Particular.ServiceControl");
@@ -29,7 +34,6 @@
     TimeSpan.FromMilliseconds(500)
 );
 
-var routing = transport.Routing();
 routing.RouteToEndpoint(typeof(PlaceOrder), "Sales");
 
 var endpointInstance = await Endpoint.Start(endpointConfiguration);
@@ -38,13 +42,13 @@
 var cancellation = new CancellationTokenSource();
 var simulatedWork = simulatedCustomers.Run(cancellation.Token);
 
-var nonInteractive = args.Length > 1 && bool.TryParse(args[1], out var isInteractive) && !isInteractive;
-var interactive = !nonInteractive;
 
 UserInterface.RunLoop("Load (ClientUI)", new Dictionary<char, (string, Action)>
 {
     ['c'] = ("toggle High/Low traffic mode", () => simulatedCustomers.ToggleTrafficMode()),
-}, writer => simulatedCustomers.WriteState(writer), interactive);
+    ['v'] = ("toggle manual mode", () => simulatedCustomers.ToggleManualMode()),
+    ['b'] = ("send message manually", () => simulatedCustomers.SendManually()),
+}, writer => simulatedCustomers.WriteState(writer));
 
 cancellation.Cancel();
 

src/ClientUI/SimulatedCustomers.cs

@@ -4,10 +4,14 @@ namespace ClientUI;
 
 class SimulatedCustomers(IEndpointInstance endpointInstance)
 {
+    private const string Letters = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
+
     public void WriteState(TextWriter output)
     {
-        var trafficMode = highTrafficMode ? "High" : "Low";
-        output.WriteLine($"{trafficMode} traffic mode - sending {rate} orders / second");
+        var trafficMode = manualMode 
+            ? "Manual sending mode" 
+            : highTrafficMode ? $"High traffic mode - sending {rate} orders / second" : $"Low traffic mode - sending {rate} orders / second";
+        output.WriteLine(trafficMode);
     }
 
     public void ToggleTrafficMode()
@@ -16,6 +20,16 @@ public void ToggleTrafficMode()
         rate = highTrafficMode ? 8 : 1;
     }
 
+    public void ToggleManualMode()
+    {
+        manualMode = !manualMode;
+    }
+
+    public void SendManually()
+    {
+        manualModeSemaphore.Release();
+    }
+
     public async Task Run(CancellationToken cancellationToken = default)
     {
         nextReset = DateTime.UtcNow.AddSeconds(1);
@@ -30,6 +44,11 @@ public async Task Run(CancellationToken cancellationToken = default)
                 nextReset = now.AddSeconds(1);
             }
 
+            if (manualMode)
+            {
+                await manualModeSemaphore.WaitAsync();
+            }
+
             await PlaceSingleOrder(cancellationToken);
             currentIntervalCount++;
 
@@ -53,17 +72,29 @@ public async Task Run(CancellationToken cancellationToken = default)
 
     Task PlaceSingleOrder(CancellationToken cancellationToken)
     {
+        var messageId = new string(Enumerable.Range(0, 4).Select(x => Letters[Random.Shared.Next(Letters.Length)]).ToArray());
+
         var placeOrderCommand = new PlaceOrder
         {
             OrderId = Guid.NewGuid().ToString()
         };
 
-        return endpointInstance.Send(placeOrderCommand, cancellationToken);
+        var sendOptions = new SendOptions();
+
+        if (manualMode)
+        {
+            sendOptions.SetHeader("MonitoringDemo.SlowMotion", "True");
+        }
+
+        sendOptions.SetMessageId(messageId);
+        return endpointInstance.Send(placeOrderCommand, sendOptions, cancellationToken);
     }
 
     bool highTrafficMode;
 
     DateTime nextReset;
     int currentIntervalCount;
     int rate = 1;
+    private bool manualMode;
+    private SemaphoreSlim manualModeSemaphore = new SemaphoreSlim(0);
 }