Refactor Basket Api eventbus using CAP

src/Services/Basket/Basket.API/Basket.API.csproj

@@ -18,6 +18,10 @@
     <PackageReference Include="AspNetCore.HealthChecks.Redis" Version="2.2.0" />
     <PackageReference Include="AspNetCore.HealthChecks.UI.Client" Version="2.2.2" />
     <PackageReference Include="Autofac.Extensions.DependencyInjection" Version="4.2.1" />
+    <PackageReference Include="DotNetCore.CAP" Version="2.5.0-preview-69210974" />
+    <PackageReference Include="DotNetCore.CAP.AzureServiceBus" Version="2.5.0-preview-69210974" />
+    <PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="2.5.0-preview-69210974" />
+    <PackageReference Include="DotNetCore.CAP.InMemoryStorage" Version="2.5.0-preview-69210974" />
     <PackageReference Include="Microsoft.ApplicationInsights.AspNetCore" Version="2.2.1" />
     <PackageReference Include="Microsoft.ApplicationInsights.DependencyCollector" Version="2.6.1" />
     <PackageReference Include="Microsoft.ApplicationInsights.Kubernetes" Version="1.0.2" />

src/Services/Basket/Basket.API/IntegrationEvents/EventHandling/OrderStartedIntegrationEventHandler.cs

@@ -6,10 +6,11 @@ using Microsoft.Extensions.Logging;
 using Serilog.Context;
 using System;
 using System.Threading.Tasks;
+using DotNetCore.CAP;
 
 namespace Basket.API.IntegrationEvents.EventHandling
 {
-    public class OrderStartedIntegrationEventHandler : IIntegrationEventHandler<OrderStartedIntegrationEvent>
+    public class OrderStartedIntegrationEventHandler : IIntegrationEventHandler<OrderStartedIntegrationEvent>, ICapSubscribe
     {
         private readonly IBasketRepository _repository;
         private readonly ILogger<OrderStartedIntegrationEventHandler> _logger;
@@ -22,6 +23,7 @@ namespace Basket.API.IntegrationEvents.EventHandling
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
+        //TODO: [CapSubscribe(nameof(OrderStartedIntegrationEvent))] 
         public async Task Handle(OrderStartedIntegrationEvent @event)
         {
             using (LogContext.PushProperty("IntegrationEventContext", $"{@event.Id}-{Program.AppName}"))
@@ -32,7 +34,4 @@ namespace Basket.API.IntegrationEvents.EventHandling
             }
         }
     }
-}
-
-
-
+}

src/Services/Basket/Basket.API/IntegrationEvents/EventHandling/ProductPriceChangedIntegrationEventHandler.cs

@@ -6,10 +6,11 @@ using Serilog.Context;
 using System;
 using System.Linq;
 using System.Threading.Tasks;
+using DotNetCore.CAP;
 
 namespace Microsoft.eShopOnContainers.Services.Basket.API.IntegrationEvents.EventHandling
 {
-    public class ProductPriceChangedIntegrationEventHandler : IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>
+    public class ProductPriceChangedIntegrationEventHandler : IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>, ICapSubscribe
     {
         private readonly ILogger<ProductPriceChangedIntegrationEventHandler> _logger;
         private readonly IBasketRepository _repository;
@@ -22,6 +23,7 @@ namespace Microsoft.eShopOnContainers.Services.Basket.API.IntegrationEvents.Even
             _repository = repository ?? throw new ArgumentNullException(nameof(repository));
         }
 
+        //TODO: [CapSubscribe(nameof(ProductPriceChangedIntegrationEvent))]
         public async Task Handle(ProductPriceChangedIntegrationEvent @event)
         {
             using (LogContext.PushProperty("IntegrationEventContext", $"{@event.Id}-{Program.AppName}"))

src/Services/Basket/Basket.API/Startup.cs

@@ -3,9 +3,7 @@ using Autofac.Extensions.DependencyInjection;
 using Basket.API.Infrastructure.Filters;
 using Basket.API.Infrastructure.Middlewares;
 using Basket.API.IntegrationEvents.EventHandling;
-using Basket.API.IntegrationEvents.Events;
 using HealthChecks.UI.Client;
-
 using Microsoft.ApplicationInsights.Extensibility;
 using Microsoft.ApplicationInsights.ServiceFabric;
 using Microsoft.AspNetCore.Authentication.JwtBearer;
@@ -14,13 +12,7 @@ using Microsoft.AspNetCore.Diagnostics.HealthChecks;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Mvc;
-using Microsoft.Azure.ServiceBus;
-using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
-using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
-using Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ;
-using Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
 using Microsoft.eShopOnContainers.Services.Basket.API.IntegrationEvents.EventHandling;
-using Microsoft.eShopOnContainers.Services.Basket.API.IntegrationEvents.Events;
 using Microsoft.eShopOnContainers.Services.Basket.API.Model;
 using Microsoft.eShopOnContainers.Services.Basket.API.Services;
 using Microsoft.Extensions.Configuration;
@@ -28,7 +20,6 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Diagnostics.HealthChecks;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using RabbitMQ.Client;
 using StackExchange.Redis;
 using Swashbuckle.AspNetCore.Swagger;
 using System;
@@ -84,49 +75,39 @@ namespace Microsoft.eShopOnContainers.Services.Basket.API
                 return ConnectionMultiplexer.Connect(configuration);
             });
 
-
-            if (Configuration.GetValue<bool>("AzureServiceBusEnabled"))
+            services.AddCap(options =>
             {
-                services.AddSingleton<IServiceBusPersisterConnection>(sp =>
+                options.UseInMemoryStorage();
+                if (Configuration.GetValue<bool>("AzureServiceBusEnabled"))
                 {
-                    var logger = sp.GetRequiredService<ILogger<DefaultServiceBusPersisterConnection>>();
-
-                    var serviceBusConnectionString = Configuration["EventBusConnection"];
-                    var serviceBusConnection = new ServiceBusConnectionStringBuilder(serviceBusConnectionString);
-
-                    return new DefaultServiceBusPersisterConnection(serviceBusConnection, logger);
-                });
-            }
-            else
-            {
-                services.AddSingleton<IRabbitMQPersistentConnection>(sp =>
+                    options.UseAzureServiceBus(Configuration["EventBusConnection"]);
+                }
+                else
                 {
-                    var logger = sp.GetRequiredService<ILogger<DefaultRabbitMQPersistentConnection>>();
-
-                    var factory = new ConnectionFactory()
+                    options.UseRabbitMQ(conf =>
                     {
-                        HostName = Configuration["EventBusConnection"]
-                    };
-
-                    if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))
-                    {
-                        factory.UserName = Configuration["EventBusUserName"];
-                    }
-
-                    if (!string.IsNullOrEmpty(Configuration["EventBusPassword"]))
-                    {
-                        factory.Password = Configuration["EventBusPassword"];
-                    }
-
-                    var retryCount = 5;
-                    if (!string.IsNullOrEmpty(Configuration["EventBusRetryCount"]))
-                    {
-                        retryCount = int.Parse(Configuration["EventBusRetryCount"]);
-                    }
+                        conf.HostName = Configuration["EventBusConnection"];
+                        if (!string.IsNullOrEmpty(Configuration["EventBusUserName"]))
+                        {
+                            conf.UserName = Configuration["EventBusUserName"];
+                        }
+                        if (!string.IsNullOrEmpty(Configuration["EventBusPassword"]))
+                        {
+                            conf.Password = Configuration["EventBusPassword"];
+                        }
+                    });
+                }
+
+                if (!string.IsNullOrEmpty(Configuration["EventBusRetryCount"]))
+                {
+                    options.FailedRetryCount = int.Parse(Configuration["EventBusRetryCount"]);
+                }
 
-                    return new DefaultRabbitMQPersistentConnection(factory, logger, retryCount);
-                });
-            }
+                if (!string.IsNullOrEmpty(Configuration["SubscriptionClientName"]))
+                {
+                    options.DefaultGroup = Configuration["SubscriptionClientName"];
+                }
+            });
 
             RegisterEventBus(services);
 
@@ -215,9 +196,6 @@ namespace Microsoft.eShopOnContainers.Services.Basket.API
                    c.OAuthClientId ("basketswaggerui");
                    c.OAuthAppName("Basket Swagger UI");
                });
-
-            ConfigureEventBus(app);
-
         }
 
         private void RegisterAppInsights(IServiceCollection services)
@@ -270,53 +248,9 @@ namespace Microsoft.eShopOnContainers.Services.Basket.API
 
         private void RegisterEventBus(IServiceCollection services)
         {
-            var subscriptionClientName = Configuration["SubscriptionClientName"];
-
-            if (Configuration.GetValue<bool>("AzureServiceBusEnabled"))
-            {
-                services.AddSingleton<IEventBus, EventBusServiceBus>(sp =>
-                {
-                    var serviceBusPersisterConnection = sp.GetRequiredService<IServiceBusPersisterConnection>();
-                    var iLifetimeScope = sp.GetRequiredService<ILifetimeScope>();
-                    var logger = sp.GetRequiredService<ILogger<EventBusServiceBus>>();
-                    var eventBusSubcriptionsManager = sp.GetRequiredService<IEventBusSubscriptionsManager>();                    
-
-                    return new EventBusServiceBus(serviceBusPersisterConnection, logger,
-                        eventBusSubcriptionsManager, subscriptionClientName, iLifetimeScope);
-                });
-            }
-            else
-            {
-                services.AddSingleton<IEventBus, EventBusRabbitMQ>(sp =>
-                {
-                    var rabbitMQPersistentConnection = sp.GetRequiredService<IRabbitMQPersistentConnection>();
-                    var iLifetimeScope = sp.GetRequiredService<ILifetimeScope>();
-                    var logger = sp.GetRequiredService<ILogger<EventBusRabbitMQ>>();
-                    var eventBusSubcriptionsManager = sp.GetRequiredService<IEventBusSubscriptionsManager>();
-
-                    var retryCount = 5;
-                    if (!string.IsNullOrEmpty(Configuration["EventBusRetryCount"]))
-                    {
-                        retryCount = int.Parse(Configuration["EventBusRetryCount"]);
-                    }
-
-                    return new EventBusRabbitMQ(rabbitMQPersistentConnection, logger, iLifetimeScope, eventBusSubcriptionsManager, subscriptionClientName, retryCount);
-                });
-            }
-
-            services.AddSingleton<IEventBusSubscriptionsManager, InMemoryEventBusSubscriptionsManager>();
-
             services.AddTransient<ProductPriceChangedIntegrationEventHandler>();
             services.AddTransient<OrderStartedIntegrationEventHandler>();
-        }
-
-        private void ConfigureEventBus(IApplicationBuilder app)
-        {
-            var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();
-
-            eventBus.Subscribe<ProductPriceChangedIntegrationEvent, ProductPriceChangedIntegrationEventHandler>();
-            eventBus.Subscribe<OrderStartedIntegrationEvent, OrderStartedIntegrationEventHandler>();
-        }        
+        }    
     }
 
     public static class CustomExtensionMethods