Merge pull request #7 from alexvaluyskiy/feature/masstransit_metrics

MassTransit metrics

.github/workflows/publish-nuget-packages.yaml

@@ -11,6 +11,6 @@ jobs:
     - uses: actions/checkout@v1
     - uses: actions/setup-dotnet@v1
       with:
-        dotnet-version: '3.0.100'
+        dotnet-version: '3.1.101'
     - run: dotnet pack prometheus-net.Contrib.sln --include-symbols -c "Release" --output "build/"
     - run: dotnet nuget push "build/*.symbols.nupkg" -k ${{ secrets.NUGET_API_KEY }} -s "https://api.nuget.org/v3/index.json" -n true

.github/workflows/run-tests.yaml

@@ -14,7 +14,6 @@ jobs:
     - uses: actions/checkout@v1
     - uses: actions/setup-dotnet@v1
       with:
-        dotnet-version: 3.0.100
+        dotnet-version: '3.1.101'
     - name: dotnet test
       run: dotnet test -c "Release"
-

Directory.Build.props

@@ -8,5 +8,7 @@
       <Company>alexvaluyskiy</Company>
       <PublishRepositoryUrl>true</PublishRepositoryUrl>
       <Version>0.7.0</Version>
+      <PackageTags>prometheus metrics</PackageTags>
+      <Description>Exposes .NET core diagnostic listeners and counters</Description>
     </PropertyGroup>
   </Project>

prometheus-net.Contrib.sln

@@ -23,7 +23,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "prometheus-net.Esquio", "sr
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{AC951721-6E15-4E0A-A15A-E268AA6DDAEA}"
 	ProjectSection(SolutionItems) = preProject
+		Directory.Build.props = Directory.Build.props
+		.github\workflows\publish-nuget-packages.yaml = .github\workflows\publish-nuget-packages.yaml
 		README.md = README.md
+		.github\workflows\run-tests.yaml = .github\workflows\run-tests.yaml
 	EndProjectSection
 EndProject
 Global

samples/WebApp/Controllers/TestController.cs

@@ -2,26 +2,27 @@
 using System.Net.Http;
 using System.Threading.Tasks;
 using MassTransit;
+using MassTransit.Courier;
 using Microsoft.AspNetCore.Mvc;
-using WebApp.Consumers;
+using WebApp.MassTransit;
 
 namespace WebApp.Controllers
 {
     [ApiController]
     [Route("[controller]")]
     public class TestController : ControllerBase
     {
-        private readonly IPublishEndpoint endpoint;
+        private readonly IBus bus;
 
-        public TestController(IPublishEndpoint endpoint)
+        public TestController(IBus bus)
         {
-            this.endpoint = endpoint;
+            this.bus = bus;
         }
 
         [HttpGet("send")]
         public async Task<IActionResult> TestSend()
         {
-            await endpoint.Publish(new TestCommand
+            await bus.Publish(new TestCommand
             {
                 Id = Guid.NewGuid()
             });
@@ -31,5 +32,36 @@ await endpoint.Publish(new TestCommand
 
             return Ok();
         }
+
+        [HttpGet("saga")]
+        public async Task<IActionResult> TestSaga()
+        {
+            var orderId = Guid.NewGuid();
+
+            await bus.Publish<SubmitOrder>(new
+            {
+                OrderId = orderId
+            });
+
+            await bus.Publish<OrderAccepted>(new
+            {
+                OrderId = orderId
+            });
+
+            return Ok();
+        }
+
+        [HttpGet("courier")]
+        public async Task<IActionResult> TestCourier()
+        {
+            var builder = new RoutingSlipBuilder(NewId.NextGuid());
+            builder.AddActivity("DownloadImage", new Uri("queue:test_courier_execute"));
+            builder.AddActivity("FilterImage", new Uri("queue:test_courier_execute"));
+            var routingSlip = builder.Build();
+
+            await bus.Execute(routingSlip);
+
+            return Ok();
+        }
     }
 }

samples/WebApp/MassTransit/OrderStateMachine.cs

@@ -0,0 +1,52 @@
+using System;
+using Automatonymous;
+
+namespace WebApp.MassTransit
+{
+    public class OrderStateMachine : MassTransitStateMachine<OrderState>
+    {
+        public OrderStateMachine()
+        {
+            InstanceState(x => x.CurrentState);
+
+            Event(() => SubmitOrder, x => x.CorrelateById(context => context.Message.OrderId));
+            Event(() => OrderAccepted, x => x.CorrelateById(context => context.Message.OrderId));
+
+            Initially(
+                When(SubmitOrder)
+                    .TransitionTo(Submitted));
+
+            During(Submitted,
+                When(OrderAccepted)
+                    .Then(context =>
+                    {
+
+                    })
+                    .TransitionTo(Accepted)
+                    .Finalize());
+        }
+
+        public Event<SubmitOrder> SubmitOrder { get; private set; }
+        public Event<OrderAccepted> OrderAccepted { get; private set; }
+
+
+        public State Submitted { get; private set; }
+        public State Accepted { get; private set; }
+    }
+
+    public class OrderState : SagaStateMachineInstance
+    {
+        public Guid CorrelationId { get; set; }
+        public string CurrentState { get; set; }
+    }
+
+    public interface SubmitOrder
+    {
+        Guid OrderId { get; }
+    }
+
+    public interface OrderAccepted
+    {
+        Guid OrderId { get; }
+    }
+}

samples/WebApp/Consumers/TestCommand.cs → samples/WebApp/MassTransit/TestCommand.cs

@@ -1,6 +1,6 @@
 using System;
 
-namespace WebApp.Consumers
+namespace WebApp.MassTransit
 {
     public class TestCommand
     {

samples/WebApp/Consumers/TestConsumer.cs → samples/WebApp/MassTransit/TestConsumer.cs

@@ -1,14 +1,11 @@
-using MassTransit;
-using StackExchange.Redis;
-using System;
-using Microsoft.Data.SqlClient;
+using System;
 using System.Threading.Tasks;
-using WebApp;
-using WebApp.Data;
+using MassTransit;
+using Microsoft.Data.SqlClient;
 using Microsoft.EntityFrameworkCore;
-using System.Transactions;
+using WebApp.Data;
 
-namespace WebApp.Consumers
+namespace WebApp.MassTransit
 {
     public class TestConsumer : IConsumer<TestCommand>
     {
@@ -23,9 +20,6 @@ public TestConsumer(SqlConnection connection, TestContext testContext)
 
         public async Task Consume(ConsumeContext<TestCommand> context)
         {
-            //var database = Startup.connection.GetDatabase();
-            //database.StringGet("test1");
-
             await testContext.TestEntities.ToListAsync();
 
             var command = connection.CreateCommand();

samples/WebApp/MassTransit/TestCourier.cs

@@ -0,0 +1,48 @@
+using System;
+using System.Threading.Tasks;
+using MassTransit.Courier;
+
+namespace WebApp.MassTransit
+{
+    public class DownloadImageActivity : IActivity<DownloadImageArguments, DownloadImageLog>
+    {
+        public async Task<ExecutionResult> Execute(ExecuteContext<DownloadImageArguments> context)
+        {
+            return default;
+        }
+
+        public async Task<CompensationResult> Compensate(CompensateContext<DownloadImageLog> context)
+        {
+            return default;
+        }
+    }
+
+    public class FilterImageActivity : IActivity<FilterImageArguments, FilterImageLog>
+    {
+        public async Task<ExecutionResult> Execute(ExecuteContext<FilterImageArguments> context)
+        {
+            return default;
+        }
+
+        public async Task<CompensationResult> Compensate(CompensateContext<FilterImageLog> context)
+        {
+            return default;
+        }
+    }
+
+    public class FilterImageLog
+    {
+    }
+
+    public class FilterImageArguments
+    {
+    }
+
+    public class DownloadImageLog
+    {
+    }
+
+    public class DownloadImageArguments
+    {
+    }
+}

samples/WebApp/Startup.cs

@@ -4,24 +4,21 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using MassTransit.AspNetCoreIntegration;
-using StackExchange.Redis;
 using Microsoft.Data.SqlClient;
 using Prometheus;
 using MassTransit;
-using WebApp.Consumers;
-using MassTransit.ActiveMqTransport;
-using MassTransit.ActiveMqTransport.Configurators;
 using System;
+using GreenPipes;
+using MassTransit.Saga;
 using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Diagnostics.HealthChecks;
 using WebApp.Data;
-using Microsoft.AspNetCore.Routing;
+using WebApp.MassTransit;
 
 namespace WebApp
 {
     public class Startup
     {
-        public static ConnectionMultiplexer connection = ConnectionMultiplexer.Connect("");
-
         public Startup(IConfiguration configuration)
         {
             Configuration = configuration;
@@ -38,21 +35,46 @@ public void ConfigureServices(IServiceCollection services)
                 options.UseSqlServer(Configuration.GetConnectionString("DefaultConnection"));
             });
 
+            services.AddSingleton<ISagaRepository<OrderState>>(provider => new InMemorySagaRepository<OrderState>());
+
             services.AddMassTransit(
-                provider => Bus.Factory.CreateUsingActiveMq(factoryConfigurator =>
+                provider => Bus.Factory.CreateUsingRabbitMq(factoryConfigurator =>
                 {
-                    var host = factoryConfigurator.Host(new ConfigurationHostSettings(new Uri("activemq://localhost:61616")));
+                    factoryConfigurator.Host(new Uri("amqp://admin:admin@localhost:5672/"));
+
                     factoryConfigurator.ReceiveEndpoint("test_events", receiveEndpointConfigurator =>
                     {
                         receiveEndpointConfigurator.Consumer<TestConsumer>(provider);
                     });
+
+                    factoryConfigurator.ReceiveEndpoint("test_saga", receiveEndpointConfigurator =>
+                    {
+                        receiveEndpointConfigurator.StateMachineSaga<OrderState>(provider);
+                    });
+
+                    factoryConfigurator.ReceiveEndpoint("test_courier_execute", receiveEndpointConfigurator =>
+                    {
+                        var compnsateUri = new Uri("queue:test_courier_compensate");
+                        receiveEndpointConfigurator.ExecuteActivityHost<DownloadImageActivity, DownloadImageArguments>(compnsateUri, provider);
+                        receiveEndpointConfigurator.ExecuteActivityHost<FilterImageActivity, FilterImageArguments>(compnsateUri, provider);
+                    });
+
+                    factoryConfigurator.ReceiveEndpoint("test_courier_compensate", receiveEndpointConfigurator =>
+                    {
+                        receiveEndpointConfigurator.CompensateActivityHost<DownloadImageActivity, DownloadImageLog>(provider);
+                    });
                 }),
                 config =>
                 {
                     config.AddConsumer<TestConsumer>();
-                });
+                    config.AddSagaStateMachine<OrderStateMachine, OrderState>();
+
+                    config.AddActivity<DownloadImageActivity, DownloadImageArguments, DownloadImageLog>();
+                    config.AddActivity<FilterImageActivity, FilterImageArguments, FilterImageLog>();
+                }, options =>  options.FailureStatus = HealthStatus.Unhealthy);
 
             services.AddPrometheusAspNetCoreMetrics();
+            services.AddPrometheusMassTransitMetrics();
 
             services.AddSingleton(provider =>
             {

samples/WebApp/WebApp.csproj

@@ -1,18 +1,20 @@
 <Project Sdk="Microsoft.NET.Sdk.Web">
 
   <PropertyGroup>
-    <TargetFramework>netcoreapp3.0</TargetFramework>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="MassTransit.ActiveMQ" Version="6.0.0" />
-    <PackageReference Include="MassTransit.AspNetCore" Version="6.0.0" />
+    <PackageReference Include="MassTransit.RabbitMQ" Version="6.0.2" />
+    <PackageReference Include="MassTransit.AspNetCore" Version="6.0.2" />
     <PackageReference Include="Microsoft.Data.SqlClient" Version="1.1.0" />
-    <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="3.0.1" />
-    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="3.0.1" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.SqlServer" Version="3.1.1" />
+    <PackageReference Include="Microsoft.EntityFrameworkCore.Tools" Version="3.1.1">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
+    </PackageReference>
     <PackageReference Include="prometheus-net" Version="3.4.0" />
     <PackageReference Include="prometheus-net.AspNetCore" Version="3.4.0" />
-    <PackageReference Include="StackExchange.Redis" Version="2.0.601" />
   </ItemGroup>
 
   <ItemGroup>

samples/WebApp/docker-compose.yml

@@ -9,19 +9,13 @@ services:
     ports:
       - 11433:1433
 
-  redis:
-    image: redis:5
-    command: ["redis-server", "--appendonly", "yes"]
-    ports:
-      - 6379:6379
-
-  activemq:
-    image: rmohr/activemq:latest
+  rabbit:
+    image: "rabbitmq:3.8.2-management"
     environment:
-      - "ACTIVEMQ_ADMIN_LOGIN=admin"
-      - "ACTIVEMQ_ADMIN_PASSWORD=admin"
-      - "ACTIVEMQ_LOGGER_LOGLEVEL=TRACE"
+      RABBITMQ_ERLANG_COOKIE: "SWQOKODSQALRPCLNMEQG"
+      RABBITMQ_DEFAULT_USER: "admin"
+      RABBITMQ_DEFAULT_PASS: "admin"
+      RABBITMQ_DEFAULT_VHOST: "/"
     ports:
-      - 8161:8161
-      - 61616:61616
-      - 61613:61613
+      - "15672:15672"
+      - "5672:5672"