UI (#206)

* Spike UI

* Fix rendering

* Delete src/MonitoringDemo/.\Marker.sln

* Use storage directory

* Multi-instance windows

* Small tweaks while looking at the code

* ProcessHandle

* Recognized commands based on help messages

* Minor todo

* Cosmetics

* Fix recognition of commands

* Regex

* Forward to all instance when uppercase

* Remove obsolete comment

* Widgets? Gadgets?

* Bring in more chaos!

* Outbox!

* Remove unused line

* Hide instances in the platform window

* Use explicit root folder for the transport

* Indent properly

* Upgrade to alpha because v1 should no longer be used (there might still be rough edges)

* Fix keys, I hope

* Fix lowercase vs uppercase keys

* Simplify process view slightly

* Command binding

* Hopefully slightly better approach for visibility handling

* Simplify process window state

* Shitty scale out and in

* More robust process handling

* Dynamic scale out of all instances except clientui

---------

Co-authored-by: Daniel Marbach <danielmarbach@users.noreply.github.com>
Co-authored-by: Szymon Pobiega <szymon.pobiega@gmail.com>

Author: danielmarbach

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>
@@ -22,6 +23,9 @@
     <Compile Include="..\MonitoringDemo\UserInterface.cs">
       <Link>UserInterface.cs</Link>
     </Compile>
+    <Compile Include="..\MonitoringDemo\DeterministicGuid.cs">
+      <Link>DeterministicGuid.cs</Link>
+    </Compile>
   </ItemGroup>
 
 </Project>

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,9 +1,17 @@
-using System.Text.Json;
+using System.Reflection;
+using System.Text.Json;
 using Billing;
 using Messages;
 using Microsoft.Extensions.DependencyInjection;
 using Shared;
 
+var instancePostfix = args.FirstOrDefault();
+
+var title = string.IsNullOrEmpty(instancePostfix) ? "Failure rate (Billing)" : $"Billing - {instancePostfix}";
+var instanceName = string.IsNullOrEmpty(instancePostfix) ? "billing" : $"billing-{instancePostfix}";
+
+var instanceId = DeterministicGuid.Create("Billing", instanceName);
+
 var endpointConfiguration = new EndpointConfiguration("Billing");
 endpointConfiguration.LimitMessageProcessingConcurrencyTo(4);
 
@@ -16,7 +24,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));
@@ -25,27 +38,30 @@
 endpointConfiguration.SendHeartbeatTo("Particular.ServiceControl");
 
 endpointConfiguration.UniquelyIdentifyRunningInstance()
-    .UsingCustomIdentifier(new Guid("1C62248E-2681-45A4-B44D-5CF93584BAD6"))
-    .UsingCustomDisplayName("original-instance");
+    .UsingCustomIdentifier(instanceId)
+    .UsingCustomDisplayName(instanceName);
 
 var metrics = endpointConfiguration.EnableMetrics();
 metrics.SendMetricDataToServiceControl(
     "Particular.Monitoring",
     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)>
+UserInterface.RunLoop(title, 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), interactive);
+}, 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();