Merge pull request #211 from Particular/wire-commands-clean

Add more simulation effects

Author: SzymonPobiega

.gitignore

@@ -12,4 +12,5 @@ MonitoringDemo.Sql/transport/
 
 **/.diagnostics/*
 .logs
-.idea
+.idea
+/otel/

src/Billing/Billing.csproj

@@ -10,6 +10,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\Messages\Messages.csproj" />
+    <ProjectReference Include="..\Shared\Shared.csproj" />
   </ItemGroup>
 
   <ItemGroup>
@@ -19,9 +20,4 @@
     <PackageReference Include="NServiceBus.Persistence.NonDurable" Version="2.*" />
   </ItemGroup>
 
-  <ItemGroup>
-    <Compile Include="..\MonitoringDemo\DeterministicGuid.cs" />
-    <Compile Include="..\MonitoringDemo\UserInterface.cs" />
-  </ItemGroup>
-
 </Project>

src/Billing/DispatchingProgressBehavior.cs

@@ -1,18 +1,20 @@
 using Messages;
+using NServiceBus;
+using Shared;
 
 namespace Billing;
 
-public class OrderPlacedHandler(SimulationEffects simulationEffects) : IHandleMessages<OrderPlaced>
+public class OrderPlacedHandler : IHandleMessages<OrderPlaced>
 {
     public async Task Handle(OrderPlaced message, IMessageHandlerContext context)
     {
-        await simulationEffects.SimulatedMessageProcessing(context.CancellationToken);
-
         var orderBilled = new OrderBilled
         {
             OrderId = message.OrderId
         };
 
-        await context.Publish(orderBilled);
+        var publishOptions = new PublishOptions();
+        publishOptions.SetMessageId(MessageIdHelper.GetHumanReadableMessageId());
+        await context.Publish(orderBilled, publishOptions);
     }
 }

src/Billing/FailureSimulation.cs

@@ -9,59 +9,75 @@
 
 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 prometheusPortString = args.Skip(1).FirstOrDefault();
 
-var endpointConfiguration = new EndpointConfiguration("Billing");
-endpointConfiguration.LimitMessageProcessingConcurrencyTo(4);
+var endpointControls = new ProcessingEndpointControls(() => PrepareEndpointConfiguration(instanceId, instanceName, prometheusPortString));
 
-var serializer = endpointConfiguration.UseSerialization<SystemJsonSerializer>();
-serializer.Options(new JsonSerializerOptions
-{
-    TypeInfoResolverChain =
-        {
-            MessagesSerializationContext.Default
-        }
-});
+var ui = new UserInterface();
+endpointControls.BindSlowProcessingDial(ui, '5', 't');
+endpointControls.BindDatabaseFailuresDial(ui, '6', 'y');
+
+endpointControls.BindDatabaseDownToggle(ui, 'f');
+endpointControls.BindDelayedRetriesToggle(ui, 'g');
+endpointControls.BindAutoThrottleToggle(ui, 'h');
 
-var transport = new LearningTransport
+endpointControls.BindFailureReceivingButton(ui, 'v');
+endpointControls.BindFailureProcessingButton(ui, 'b');
+endpointControls.BindFailureDispatchingButton(ui, 'n');
+
+if (prometheusPortString != null)
 {
-    StorageDirectory = Path.Combine(Directory.GetParent(Assembly.GetExecutingAssembly().Location)!.Parent!.FullName, ".learningtransport"),
-    TransportTransactionMode = TransportTransactionMode.ReceiveOnly
-};
-endpointConfiguration.UseTransport(transport);
+    OpenTelemetryUtils.ConfigureOpenTelemetry("Billing", instanceId.ToString(), int.Parse(prometheusPortString));
+}
 
-endpointConfiguration.Recoverability()
-    .Delayed(delayed => delayed.NumberOfRetries(0));
+endpointControls.Start();
 
-endpointConfiguration.AuditProcessedMessagesTo("audit");
-endpointConfiguration.SendHeartbeatTo("Particular.ServiceControl");
+ui.RunLoop(title);
 
-endpointConfiguration.UniquelyIdentifyRunningInstance()
-    .UsingCustomIdentifier(instanceId)
-    .UsingCustomDisplayName(instanceName);
+await endpointControls.StopEndpoint();
 
-var metrics = endpointConfiguration.EnableMetrics();
-metrics.SendMetricDataToServiceControl(
-    "Particular.Monitoring",
-    TimeSpan.FromMilliseconds(500)
-);
+EndpointConfiguration PrepareEndpointConfiguration(Guid guid, string s, string? prometheusPortString1)
+{
+    var endpointConfiguration1 = new EndpointConfiguration("Billing");
+    endpointConfiguration1.LimitMessageProcessingConcurrencyTo(4);
 
-endpointConfiguration.UsePersistence<NonDurablePersistence>();
-endpointConfiguration.EnableOutbox();
+    var serializer = endpointConfiguration1.UseSerialization<SystemJsonSerializer>();
+    serializer.Options(new JsonSerializerOptions
+    {
+        TypeInfoResolverChain =
+        {
+            MessagesSerializationContext.Default
+        }
+    });
 
-var failureSimulation = new FailureSimulation();
-failureSimulation.Register(endpointConfiguration);
+    var transport = new LearningTransport
+    {
+        StorageDirectory = Path.Combine(Directory.GetParent(Assembly.GetExecutingAssembly().Location)!.Parent!.FullName, ".learningtransport"),
+        TransportTransactionMode = TransportTransactionMode.ReceiveOnly
+    };
+    endpointConfiguration1.UseTransport(transport);
 
-var simulationEffects = new SimulationEffects();
-endpointConfiguration.RegisterComponents(cc => cc.AddSingleton(simulationEffects));
+    endpointConfiguration1.Recoverability()
+        .Delayed(delayed => delayed.NumberOfRetries(0));
 
-var endpointInstance = await Endpoint.Start(endpointConfiguration);
+    endpointConfiguration1.AuditProcessedMessagesTo("audit");
+    endpointConfiguration1.SendHeartbeatTo("Particular.ServiceControl");
 
-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));
+    endpointConfiguration1.UniquelyIdentifyRunningInstance()
+        .UsingCustomIdentifier(guid)
+        .UsingCustomDisplayName(s);
+
+    var metrics = endpointConfiguration1.EnableMetrics();
+    metrics.SendMetricDataToServiceControl(
+        "Particular.Monitoring",
+        TimeSpan.FromMilliseconds(500)
+    );
+
+    endpointConfiguration1.UsePersistence<NonDurablePersistence>();
+    endpointConfiguration1.EnableOutbox();
+
+    endpointConfiguration1.EnableOpenTelemetry();
 
-await endpointInstance.Stop();
+    return endpointConfiguration1;
+}