Refactor Marketing Api eventbus using CAP

src/Services/Marketing/Marketing.API/IntegrationEvents/Events/UserLocationUpdatedIntegrationEvent.cs

@@ -1,10 +1,9 @@
 namespace Microsoft.eShopOnContainers.Services.Marketing.API.IntegrationEvents.Events
 {
     using Marketing.API.Model;
-    using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
     using System.Collections.Generic;
 
-    public class UserLocationUpdatedIntegrationEvent : IntegrationEvent
+    public class UserLocationUpdatedIntegrationEvent 
     {
         public string UserId { get; set; }
 

src/Services/Marketing/Marketing.API/IntegrationEvents/Handlers/UserLocationUpdatedIntegrationEventHandler.cs

@@ -1,8 +1,9 @@
-namespace Microsoft.eShopOnContainers.Services.Marketing.API.IntegrationEvents.Handlers
+using DotNetCore.CAP;
+
+namespace Microsoft.eShopOnContainers.Services.Marketing.API.IntegrationEvents.Handlers
 {
     using Marketing.API.IntegrationEvents.Events;
     using Marketing.API.Model;
-    using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
     using Microsoft.eShopOnContainers.Services.Marketing.API.Infrastructure.Repositories;
     using Microsoft.Extensions.Logging;
     using Serilog.Context;
@@ -10,8 +11,7 @@
     using System.Collections.Generic;
     using System.Threading.Tasks;
 
-    public class UserLocationUpdatedIntegrationEventHandler 
-        : IIntegrationEventHandler<UserLocationUpdatedIntegrationEvent>
+    public class UserLocationUpdatedIntegrationEventHandler : ICapSubscribe
     {
         private readonly IMarketingDataRepository _marketingDataRepository;
         private readonly ILogger<UserLocationUpdatedIntegrationEventHandler> _logger;
@@ -24,11 +24,12 @@
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
         }
 
+        //TODO: [CapSubscribe(nameof(UserLocationUpdatedIntegrationEvent))]
         public async Task Handle(UserLocationUpdatedIntegrationEvent @event)
         {
-            using (LogContext.PushProperty("IntegrationEventContext", $"{@event.Id}-{Program.AppName}"))
+            using (LogContext.PushProperty("IntegrationEventContext", $"{Program.AppName}"))
             {
-                _logger.LogInformation("----- Handling integration event: {IntegrationEventId} at {AppName} - ({@IntegrationEvent})", @event.Id, Program.AppName, @event);
+                _logger.LogInformation("----- Handling integration event: {AppName} - ({@IntegrationEvent})", Program.AppName, @event);
 
                 var userMarketingData = await _marketingDataRepository.GetAsync(@event.UserId);
                 userMarketingData = userMarketingData ??

src/Services/Marketing/Marketing.API/Marketing.API.csproj

@@ -27,6 +27,10 @@
     <PackageReference Include="AspNetCore.HealthChecks.SqlServer" 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.MongoDB" 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" />
@@ -36,10 +40,10 @@
     <PackageReference Include="Microsoft.AspNetCore.HealthChecks" Version="1.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.AzureKeyVault" Version="2.2.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.AzureAppServices" Version="2.2.0" />
-    <PackageReference Include="mongocsharpdriver" Version="2.5.0" />
-    <PackageReference Include="MongoDB.Bson" Version="2.5.0" />
-    <PackageReference Include="MongoDB.Driver" Version="2.5.0" />
-    <PackageReference Include="MongoDB.Driver.Core" Version="2.5.0" />
+    <PackageReference Include="mongocsharpdriver" Version="2.7.2" />
+    <PackageReference Include="MongoDB.Bson" Version="2.7.2" />
+    <PackageReference Include="MongoDB.Driver" Version="2.7.2" />
+    <PackageReference Include="MongoDB.Driver.Core" Version="2.7.2" />
     <PackageReference Include="Serilog.AspNetCore" Version="2.1.1" />
     <PackageReference Include="Serilog.Enrichers.Environment" Version="2.1.2" />
     <PackageReference Include="Serilog.Settings.Configuration" Version="3.0.1" />
@@ -48,8 +52,6 @@
     <PackageReference Include="Swashbuckle.AspNetCore" Version="3.0.0" />
   </ItemGroup>
   <ItemGroup>
-    <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusRabbitMQ\EventBusRabbitMQ.csproj" />
-    <ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusServiceBus\EventBusServiceBus.csproj" />
     <ProjectReference Include="..\..\..\BuildingBlocks\WebHostCustomization\WebHost.Customization\WebHost.Customization.csproj" />
   </ItemGroup>
 

src/Services/Marketing/Marketing.API/Startup.cs

@@ -5,11 +5,6 @@
     using AspNetCore.Http;
     using Autofac;
     using Autofac.Extensions.DependencyInjection;
-    using Azure.ServiceBus;
-    using BuildingBlocks.EventBus;
-    using BuildingBlocks.EventBus.Abstractions;
-    using BuildingBlocks.EventBusRabbitMQ;
-    using BuildingBlocks.EventBusServiceBus;
     using EntityFrameworkCore;
     using Extensions.Configuration;
     using Extensions.DependencyInjection;
@@ -19,7 +14,6 @@
     using Infrastructure.Filters;
     using Infrastructure.Repositories;
     using Infrastructure.Services;
-    using IntegrationEvents.Events;
     using Marketing.API.IntegrationEvents.Handlers;
     using Microsoft.ApplicationInsights.Extensibility;
     using Microsoft.ApplicationInsights.ServiceFabric;
@@ -81,49 +75,48 @@
                 //Check Client vs. Server evaluation: https://docs.microsoft.com/en-us/ef/core/querying/client-eval
             });
 
-            if (Configuration.GetValue<bool>("AzureServiceBusEnabled"))
+            services.AddTransient<UserLocationUpdatedIntegrationEventHandler>();
+ 
+            services.AddCap(options =>
             {
-                services.AddSingleton<IServiceBusPersisterConnection>(sp =>
+                // using MongoDB as the event storage
+                options.UseMongoDB(configure =>
                 {
-                    var logger = sp.GetRequiredService<ILogger<DefaultServiceBusPersisterConnection>>();
-
-                    var serviceBusConnectionString = Configuration["EventBusConnection"];
-                    var serviceBusConnection = new ServiceBusConnectionStringBuilder(serviceBusConnectionString);
-
-                    return new DefaultServiceBusPersisterConnection(serviceBusConnection, logger);
+                    configure.DatabaseConnection = Configuration["MongoConnectionString"];
+                    configure.DatabaseName = Configuration["MongoDatabase"];
                 });
-            }
-            else
-            {
-                services.AddSingleton<IRabbitMQPersistentConnection>(sp =>
-                {
-                    var logger = sp.GetRequiredService<ILogger<DefaultRabbitMQPersistentConnection>>();
 
-                    var factory = new ConnectionFactory()
-                    {
-                        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"]))
+                if (Configuration.GetValue<bool>("AzureServiceBusEnabled"))
+                {
+                    options.UseAzureServiceBus(Configuration["EventBusConnection"]);
+                }
+                else
+                {
+                    options.UseRabbitMQ(conf =>
                     {
-                        retryCount = int.Parse(Configuration["EventBusRetryCount"]);
-                    }
-
-                    return new DefaultRabbitMQPersistentConnection(factory, logger, retryCount);
-                });
-            }            
+                        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"]);
+                }
 
+                if (!string.IsNullOrEmpty(Configuration["SubscriptionClientName"]))
+                {
+                    options.DefaultGroup = Configuration["SubscriptionClientName"];
+                }
+            });
+             
             // Add framework services.
             services.AddSwaggerGen(options =>
             {
@@ -159,9 +152,7 @@
                     .AllowAnyMethod()
                     .AllowAnyHeader()
                     .AllowCredentials());
-            });
-
-            RegisterEventBus(services);
+            }); 
 
             services.AddTransient<IMarketingDataRepository, MarketingDataRepository>();
             services.AddSingleton<IHttpContextAccessor, HttpContextAccessor>();
@@ -212,9 +203,7 @@
                    c.SwaggerEndpoint($"{ (!string.IsNullOrEmpty(pathBase) ? pathBase : string.Empty) }/swagger/v1/swagger.json", "Marketing.API V1");
                    c.OAuthClientId("marketingswaggerui");
                    c.OAuthAppName("Marketing Swagger UI");
-               });            
-
-            ConfigureEventBus(app);
+               });
         }
 
         private void RegisterAppInsights(IServiceCollection services)
@@ -253,52 +242,6 @@
                 });
         }
 
-        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<UserLocationUpdatedIntegrationEventHandler>();
-        }
-
-        private void ConfigureEventBus(IApplicationBuilder app)
-        {
-            var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();
-            eventBus.Subscribe<UserLocationUpdatedIntegrationEvent, UserLocationUpdatedIntegrationEventHandler>();
-        }
-
         protected virtual void ConfigureAuth(IApplicationBuilder app)
         {
             if (Configuration.GetValue<bool>("UseLoadTest"))