Merge pull request #78 from Particular/ui-features

Author: johnsimons

.github/workflows/ci.yml

@@ -27,3 +27,13 @@ jobs:
           dotnet-version: 8.0.x
       - name: Build
         run: dotnet build src --configuration Release -graph
+      - name: Set up Node.js
+        uses: actions/[email protected]
+        with:
+          node-version: 21.6.x
+      - name: install frontend packages
+        run: npm install
+        working-directory: src/frontend
+      - name: build frontend
+        run: npm run build
+        working-directory: src/frontend

README.md

@@ -1,22 +1,23 @@
-# MassTransitShowcaseDemo
+# The Particular Platform for MassTransit
 
+## Launching the Showcase
 
-## Launching the Particular Platform for MassTransit
-To run the code sample you have 3 options in terms of transports:
+To run the code sample you have 2 options in terms of transports:
 
 ## **RabbitMQ**
 
->[!NOTE]
-The RabbitMQ Broker is started as part of the docker compose process
->
+> [!NOTE]
+> The RabbitMQ Broker is started as part of the docker compose process
 
 Run docker command below from the `src` folder in a CLI
+
 ```cmd
 docker compose -f docker-compose-base.yml -f compose-rabbitmq.yml --env-file rabbit.env up -d
 ```
+
 ## **Azure ServiceBus**
 
-Configure the access to your Azure Service Bus namespace by editing the variables in `src/asb.env`  
+Configure the access to your Azure Service Bus namespace by editing the variables in `src/asb.env`
 
 ```env
 CONNECTIONSTRING="Endpoint=sb://[NAMESPACE].servicebus.windows.net/;SharedAccessKeyName=[KEYNAME];SharedAccessKey=[KEY]"
@@ -30,16 +31,14 @@ docker compose -f docker-compose-base.yml -f compose-azure.yml --env-file asb.en
 
 ## Running the code
 
-
->[!WARNING]
+> [!WARNING]
 > When using Visual Studio, make sure you have "Enable Multi-Project Launch profiles" setting on. Allow Visual Studio 2024 "multi-launch" so you can easily select the profile you want to run.
 >
 > It can be activated by accessing the Tools menu -> Manage preview features- Enable Multi-Project Launch profiles.
->
 
-After opening the solutions (from Visual Studio or Rider), choose one of the run profiles  that matches the transport configured previously:
+After opening the solutions (from Visual Studio or Rider), choose one of the run profiles that matches the transport configured previously:
+
 - `RabbitMQ`
-- `AmazonSQS`
 - `Azure Service Bus`
 
 Run the solution to start the demo.
@@ -49,5 +48,4 @@ Navigate to http://localhost:9090/ to see the UI
 
 The sample consists of 4 console applications hosting MassTransit message producers and consumers that implement a simplified order processing logic from an e-commerce system.
 
-
 ![System Overview](diagram.svg "width=680")

src/Billing/Billing.csproj

@@ -13,6 +13,10 @@
     <ProjectReference Include="..\Messages\Messages.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <FrameworkReference Include="Microsoft.AspNetCore.App" />
+  </ItemGroup>
+  
   <ItemGroup>
     <PackageReference Include="MassTransit.Azure.ServiceBus.Core" Version="8.3.0" />
     <PackageReference Include="MassTransit.RabbitMQ" Version="8.3.0" />

src/Billing/BillingHub.cs

@@ -0,0 +1,19 @@
+namespace Billing
+{
+    using Microsoft.AspNetCore.SignalR;
+
+    public class BillingHub(SimulationEffects simulationEffects) : Hub
+    {
+        public override async Task OnConnectedAsync()
+        {
+            await Clients.Caller.SendAsync("SyncValues", simulationEffects.MessagesProcessed, simulationEffects.MessagesErrored, simulationEffects.ShouldFailRetries);
+            await base.OnConnectedAsync();
+        }
+
+        public async Task SetFailRetries(bool shouldFailRetries)
+        {
+            simulationEffects.ShouldFailRetries = shouldFailRetries;
+            await Clients.Caller.SendAsync("SyncValues", simulationEffects.MessagesProcessed, simulationEffects.MessagesErrored, simulationEffects.ShouldFailRetries);
+        }
+    }
+}

src/Billing/ConsoleBackgroundService.cs

@@ -4,20 +4,33 @@
 using Helper;
 using MassTransit;
 using Messages;
+using Microsoft.AspNetCore.SignalR;
 
-public class BillOrderConsumer(SimulationEffects simulationEffects) : IConsumer<OrderPlaced>
+public class BillOrderConsumer(SimulationEffects simulationEffects, IHubContext<BillingHub> billingHub) : IConsumer<OrderPlaced>
 {
     public async Task Consume(ConsumeContext<OrderPlaced> context)
     {
-        await simulationEffects.SimulatedMessageProcessing(context.CancellationToken);
+        await billingHub.Clients.All.SendAsync("ProcessingMessage", context.Message, context.CancellationToken);
+        try
+        {
+            await simulationEffects.SimulateBillingProcessing(context);
+        }
+        catch (OperationCanceledException) when (context.CancellationToken.IsCancellationRequested) { throw; }
+        catch
+        {
+            var messageViewId = DeterministicGuid.GetTheViewId(context.MessageId.ToString(), context.ReceiveContext.InputAddress.ToString());
+
+            await billingHub.Clients.All.SendAsync("MessageError", context.Message, context.MessageId, messageViewId, context.CancellationToken);
+            throw;
+        }
 
         var orderBilled = new OrderBilled
         {
-            OrderId = context.Message.OrderId
+            OrderId = context.Message.OrderId,
+            Contents = context.Message.Contents
         };
 
         await context.Publish(orderBilled);
-
-        await ConsoleHelper.WriteMessageProcessed(context.SentTime ?? DateTime.UtcNow);
+        await billingHub.Clients.All.SendAsync("OrderBilled", orderBilled, context.CancellationToken);
     }
 }

src/Billing/Consumers/BillOrderConsumer.cs

@@ -6,6 +6,8 @@
 using System.Text;
 using Microsoft.Extensions.Logging;
 using System;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Builder;
 
 class Program
 {
@@ -14,16 +16,31 @@ public static IHostBuilder CreateHostBuilder(string[] args)
         Console.OutputEncoding = Encoding.UTF8;
         var host = Host.CreateDefaultBuilder(args)
             .ConfigureLogging(cfg => cfg.SetMinimumLevel(LogLevel.Warning))
-            .ConfigureServices((hostContext, services) =>
+            .ConfigureWebHostDefaults(webBuilder =>
             {
-                services.AddMassTransit(x =>
+                webBuilder.ConfigureServices(services =>
                 {
-                    x.AddConsumers(Assembly.GetExecutingAssembly());
-                    x.SetupTransport(args);
-                });
+                    services.AddMassTransit(x =>
+                    {
+                        x.SetKebabCaseEndpointNameFormatter();
+                        x.AddConsumers(Assembly.GetExecutingAssembly());
+                        x.SetupTransport(args);
+                    });
 
-                services.AddSingleton<SimulationEffects>();
-                services.AddHostedService<ConsoleBackgroundService>();
+                    services.AddCors();
+                    services.AddSignalR(options => { options.EnableDetailedErrors = true; });
+                    services.AddSingleton<SimulationEffects>();
+                });
+                webBuilder.UseUrls("http://*:5002");
+                webBuilder.Configure(app =>
+                {
+                    app.UseCors(options => options.AllowAnyHeader().AllowAnyMethod().WithOrigins("http://localhost:61335").AllowCredentials());
+                    app.UseRouting();
+                    app.UseEndpoints(endpoints =>
+                    {
+                        endpoints.MapHub<BillingHub>("/billingHub");
+                    });
+                });
             });
 
         return host;
@@ -36,4 +53,4 @@ static async Task Main(string[] args)
         var host = CreateHostBuilder(args).Build();
         await host.RunAsync();
     }
-}
+}

src/Billing/Program.cs

@@ -1,28 +1,42 @@
 namespace Billing;
 
-public class SimulationEffects
+using MassTransit;
+using Messages;
+using Microsoft.AspNetCore.SignalR;
+
+public class SimulationEffects(IHubContext<BillingHub> billingHub)
 {
-    public void IncreaseFailureRate() => failureRate = Math.Min(1, failureRate + FailureRateIncrement);
+    int messagesErrored = 0;
+    int messagesProcessed = 0;
 
-    public void DecreaseFailureRate() => failureRate = Math.Max(0, failureRate - FailureRateIncrement);
+    public int MessagesProcessed { get => messagesProcessed; private set => messagesProcessed = value; }
+    public int MessagesErrored { get => messagesErrored; private set => messagesErrored = value; }
+    public bool ShouldFailRetries { get; set; } = false;
 
-    public void WriteState(TextWriter output) => output.WriteLine("Failure rate: {0:P0}", failureRate);
 
-    public async Task SimulatedMessageProcessing(CancellationToken cancellationToken = default)
+    public async Task SimulateBillingProcessing(ConsumeContext<OrderPlaced> context)
     {
-        await Task.Delay(200, cancellationToken);
+        try
+        {
+            context.TryGetHeader("FailOn", out string failOn);
+            //Retries leave ServiceControl headers on the ReceiveContext. Choosing one at random here...
+            var isRetry = context.ReceiveContext.TransportHeaders.TryGetHeader("ServiceControl.RetryTo", out var _);
+            if (isRetry)
+            {
+                await billingHub.Clients.All.SendAsync("RetryAttempted");
+            }
+            if (Enum.TryParse(failOn, out Consumers endpointName) && endpointName == Consumers.Billing
+                    && (!isRetry || ShouldFailRetries))
+            {
+                Interlocked.Increment(ref messagesErrored);
+                throw new Exception($"A simulated failure occurred in Billing, OrderId: {context.Message.OrderId}, Contents: {string.Join(", ", context.Message.Contents)}");
+            }
 
-        if (Random.Shared.NextDouble() < failureRate)
+            Interlocked.Increment(ref messagesProcessed);
+        }
+        finally
         {
-            throw new Exception("BOOM! A failure occurred");
+            await billingHub.Clients.All.SendAsync("SyncValues", MessagesProcessed, MessagesErrored, ShouldFailRetries, context.CancellationToken);
         }
     }
-
-    double failureRate;
-    const double FailureRateIncrement = 0.1;
-
-    public void Reset()
-    {
-        failureRate = 0;
-    }
 }

src/Billing/SimulationEffects.cs

@@ -0,0 +1,36 @@
+namespace ClientUI
+{
+    using Messages;
+    using Microsoft.AspNetCore.SignalR;
+
+    public class ClientHub(SimulatedCustomers simulatedCustomers) : Hub
+    {
+        public override async Task OnConnectedAsync()
+        {
+            await Clients.Caller.SendAsync("Initialise", simulatedCustomers.OrdersPlaced, Enum.GetValues<Consumers>().Select(x => x.ToString()));
+            await base.OnConnectedAsync();
+        }
+
+        public async Task CreateOrder(float requestCount, string failOn)
+        {
+            for (int i = 0; i < (int)requestCount; i++)
+            {
+                var order = await simulatedCustomers.PlaceSingleOrder(failOn, Context.ConnectionAborted);
+                await Clients.All.SendAsync("OrderRequested", order, simulatedCustomers.OrdersPlaced, Context.ConnectionAborted);
+            }
+        }
+
+        public async Task RunScenario()
+        {
+            foreach (var failOn in Enum.GetValues<Consumers>())
+            {
+                for (int i = 0; i < 3; i++)
+                {
+                    var order = await simulatedCustomers.PlaceSingleOrder(failOn.ToString(), Context.ConnectionAborted);
+                    await Clients.All.SendAsync("OrderRequested", order, simulatedCustomers.OrdersPlaced, Context.ConnectionAborted);
+                }
+            }
+        }
+    }
+
+}

src/ClientUI/ClientHub.cs

@@ -14,7 +14,11 @@
     <PackageReference Include="MassTransit.AmazonSQS" Version="8.3.0" />
     <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.1" />
   </ItemGroup>
-
+  
+  <ItemGroup>
+    <FrameworkReference Include="Microsoft.AspNetCore.App" />
+  </ItemGroup>
+  
   <ItemGroup>
     <ProjectReference Include="..\Messages\Messages.csproj" />
     <ProjectReference Include="..\Helper\Helper.csproj" />