ayan.poddar/eShopOnContainers
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Refactor Webhooks Api eventbus using CAP

Browse Source
This commit is contained in:
Savorboard 2019-03-14 14:39:56 +08:00
parent cf44ba2fd8
commit 60e3d9ba9c
8 changed files with 64 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/Services/Webhooks/Webhooks.API/IntegrationEvents/OrderStatusChangedToPaidIntegrationEvent.cs
View File

@ -1,12 +1,8 @@
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using System.Collections.Generic;
namespace Webhooks.API.IntegrationEvents
{
public class OrderStatusChangedToPaidIntegrationEvent : IntegrationEvent
public class OrderStatusChangedToPaidIntegrationEvent
{
public int OrderId { get; }
public IEnumerable<OrderStockItem> OrderStockItems { get; }

9
src/Services/Webhooks/Webhooks.API/IntegrationEvents/OrderStatusChangedToPaidIntegrationEventHandler.cs
View File

@ -1,15 +1,13 @@
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Linq;
using System.Threading.Tasks;
using DotNetCore.CAP;
using Webhooks.API.Model;
using Webhooks.API.Services;
using Microsoft.Extensions.Logging;
namespace Webhooks.API.IntegrationEvents
{
public class OrderStatusChangedToPaidIntegrationEventHandler : IIntegrationEventHandler<OrderStatusChangedToPaidIntegrationEvent>
public class OrderStatusChangedToPaidIntegrationEventHandler : ICapSubscribe
{
private readonly IWebhooksRetriever _retriever;
private readonly IWebhooksSender _sender;
@ -21,6 +19,7 @@ namespace Webhooks.API.IntegrationEvents
_logger = logger;
}
//TODO [CapSubscribe(nameof(OrderStatusChangedToPaidIntegrationEvent))]
public async Task Handle(OrderStatusChangedToPaidIntegrationEvent @event)
{
var subscriptions = await _retriever.GetSubscriptionsOfType(WebhookType.OrderPaid);

10
src/Services/Webhooks/Webhooks.API/IntegrationEvents/OrderStatusChangedToShippedIntegrationEvent.cs
View File

@ -1,12 +1,6 @@
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
namespace Webhooks.API.IntegrationEvents
namespace Webhooks.API.IntegrationEvents
{
public class OrderStatusChangedToShippedIntegrationEvent : IntegrationEvent
public class OrderStatusChangedToShippedIntegrationEvent
{
public int OrderId { get; private set; }
public string OrderStatus { get; private set; }

9
src/Services/Webhooks/Webhooks.API/IntegrationEvents/OrderStatusChangedToShippedIntegrationEventHandler.cs
View File

@ -1,15 +1,13 @@
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Linq;
using System.Threading.Tasks;
using DotNetCore.CAP;
using Webhooks.API.Model;
using Webhooks.API.Services;
using Microsoft.Extensions.Logging;
namespace Webhooks.API.IntegrationEvents
{
public class OrderStatusChangedToShippedIntegrationEventHandler : IIntegrationEventHandler<OrderStatusChangedToShippedIntegrationEvent>
public class OrderStatusChangedToShippedIntegrationEventHandler :ICapSubscribe
{
private readonly IWebhooksRetriever _retriever;
private readonly IWebhooksSender _sender;
@ -21,6 +19,7 @@ namespace Webhooks.API.IntegrationEvents
_logger = logger;
}
//TODO [CapSubscribe(nameof(OrderStatusChangedToShippedIntegrationEvent)))]
public async Task Handle(OrderStatusChangedToShippedIntegrationEvent @event)
{
var subscriptions = await _retriever.GetSubscriptionsOfType(WebhookType.OrderShipped);

10
src/Services/Webhooks/Webhooks.API/IntegrationEvents/ProductPriceChangedIntegrationEvent.cs
View File

@ -1,12 +1,6 @@
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
namespace Webhooks.API.IntegrationEvents
namespace Webhooks.API.IntegrationEvents
{
public class ProductPriceChangedIntegrationEvent : IntegrationEvent
public class ProductPriceChangedIntegrationEvent
{
public int ProductId { get; private set; }

14
src/Services/Webhooks/Webhooks.API/IntegrationEvents/ProductPriceChangedIntegrationEventHandler.cs
View File

@ -1,16 +1,14 @@
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks;
using DotNetCore.CAP;
namespace Webhooks.API.IntegrationEvents
{
public class ProductPriceChangedIntegrationEventHandler : IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>
public class ProductPriceChangedIntegrationEventHandler : ICapSubscribe
{
public async Task Handle(ProductPriceChangedIntegrationEvent @event)
//TODO [CapSubscribe(nameof(ProductPriceChangedIntegrationEvent))]
public Task Handle(ProductPriceChangedIntegrationEvent @event)
{
int i = 0;
return Task.CompletedTask;
}
}
}

131
src/Services/Webhooks/Webhooks.API/Startup.cs
View File

@ -1,11 +1,8 @@
using System;
using System.Collections.Generic;
using System.Data.Common;
using System.IdentityModel.Tokens.Jwt;
using System.Linq;
using System.Reflection;
using System.Threading;
using System.Threading.Tasks;
using Autofac;
using Autofac.Extensions.DependencyInjection;
using HealthChecks.UI.Client;
@ -14,22 +11,14 @@ using Microsoft.ApplicationInsights.ServiceFabric;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Diagnostics.HealthChecks;
using Microsoft.AspNetCore.Hosting;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.ServiceBus;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Diagnostics;
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus;
using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Abstractions;
using Microsoft.eShopOnContainers.BuildingBlocks.EventBusRabbitMQ;
using Microsoft.eShopOnContainers.BuildingBlocks.EventBusServiceBus;
using Microsoft.eShopOnContainers.BuildingBlocks.IntegrationEventLogEF.Services;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Diagnostics.HealthChecks;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using Swashbuckle.AspNetCore.Swagger;
using Webhooks.API.Infrastructure;
using Webhooks.API.IntegrationEvents;
@ -56,7 +45,7 @@ namespace Webhooks.API
.AddSwagger(Configuration)
.AddCustomHealthCheck(Configuration)
.AddHttpClientServices(Configuration)
.AddIntegrationServices(Configuration)
.AddIntegrationEventHandler()
.AddEventBus(Configuration)
.AddCustomAuthentication(Configuration)
.AddSingleton<IHttpContextAccessor, HttpContextAccessor>()
@ -106,8 +95,6 @@ namespace Webhooks.API
c.OAuthClientId("webhooksswaggerui");
c.OAuthAppName("WebHooks Service Swagger UI");
});
ConfigureEventBus(app);
}
protected virtual void ConfigureAuth(IApplicationBuilder app)
@ -121,14 +108,6 @@ namespace Webhooks.API
app.UseAuthentication();
}
protected virtual void ConfigureEventBus(IApplicationBuilder app)
{
var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();
eventBus.Subscribe<ProductPriceChangedIntegrationEvent, ProductPriceChangedIntegrationEventHandler>();
eventBus.Subscribe<OrderStatusChangedToShippedIntegrationEvent, OrderStatusChangedToShippedIntegrationEventHandler>();
eventBus.Subscribe<OrderStatusChangedToPaidIntegrationEvent, OrderStatusChangedToPaidIntegrationEventHandler>();
}
}
static class CustomExtensionMethods
@ -226,47 +205,51 @@ namespace Webhooks.API
return services;
}
public static IServiceCollection AddEventBus(this IServiceCollection services, IConfiguration configuration)
{
var subscriptionClientName = configuration["SubscriptionClientName"];
services.AddCap(options =>
{
options.UseInMemoryStorage();
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);
});
options.UseAzureServiceBus(configuration["EventBusConnection"]);
}
else
{
services.AddSingleton<IEventBus, EventBusRabbitMQ>(sp =>
options.UseRabbitMQ(conf =>
{
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"]))
conf.HostName = configuration["EventBusConnection"];
if (!string.IsNullOrEmpty(configuration["EventBusUserName"]))
{
retryCount = int.Parse(configuration["EventBusRetryCount"]);
conf.UserName = configuration["EventBusUserName"];
}
if (!string.IsNullOrEmpty(configuration["EventBusPassword"]))
{
conf.Password = configuration["EventBusPassword"];
}
return new EventBusRabbitMQ(rabbitMQPersistentConnection, logger, iLifetimeScope, eventBusSubcriptionsManager, subscriptionClientName, retryCount);
});
}
services.AddSingleton<IEventBusSubscriptionsManager, InMemoryEventBusSubscriptionsManager>();
services.AddTransient<ProductPriceChangedIntegrationEventHandler>();
services.AddTransient<OrderStatusChangedToShippedIntegrationEventHandler>();
services.AddTransient<OrderStatusChangedToPaidIntegrationEventHandler>();
if (!string.IsNullOrEmpty(configuration["EventBusRetryCount"]))
{
options.FailedRetryCount = int.Parse(configuration["EventBusRetryCount"]);
}
if (!string.IsNullOrEmpty(configuration["SubscriptionClientName"]))
{
options.DefaultGroup = configuration["SubscriptionClientName"];
}
});
return services;
}
public static IServiceCollection AddIntegrationEventHandler(this IServiceCollection services)
{
services.AddTransient<ProductPriceChangedIntegrationEventHandler>(); //Subscribe for ProductPriceChangedIntegrationEvent
services.AddTransient<OrderStatusChangedToShippedIntegrationEventHandler>(); //Subscribe for OrderStatusChangedToShippedIntegrationEvent
services.AddTransient<OrderStatusChangedToPaidIntegrationEventHandler>(); //Subscribe for OrderStatusChangedToPaidIntegrationEvent
return services;
}
@ -298,54 +281,6 @@ namespace Webhooks.API
return services;
}
public static IServiceCollection AddIntegrationServices(this IServiceCollection services, IConfiguration configuration)
{
services.AddTransient<Func<DbConnection, IIntegrationEventLogService>>(
sp => (DbConnection c) => new IntegrationEventLogService(c));
if (configuration.GetValue<bool>("AzureServiceBusEnabled"))
{
services.AddSingleton<IServiceBusPersisterConnection>(sp =>
{
var logger = sp.GetRequiredService<ILogger<DefaultServiceBusPersisterConnection>>();
var serviceBusConnection = new ServiceBusConnectionStringBuilder(configuration["EventBusConnection"]);
return new DefaultServiceBusPersisterConnection(serviceBusConnection, logger);
});
}
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"]))
{
retryCount = int.Parse(configuration["EventBusRetryCount"]);
}
return new DefaultRabbitMQPersistentConnection(factory, logger, retryCount);
});
}
return services;
}
public static IServiceCollection AddCustomAuthentication(this IServiceCollection services, IConfiguration configuration)
{
// prevent from mapping "sub" claim to nameidentifier.

8
src/Services/Webhooks/Webhooks.API/Webhooks.API.csproj
View File

@ -11,6 +11,10 @@
<PackageReference Include="Microsoft.AspNetCore.Razor.Design" Version="2.2.0" PrivateAssets="All" />
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.0.2105168" />
<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" />
@ -22,10 +26,6 @@
<PackageReference Include="AspNetCore.HealthChecks.SqlServer" Version="2.2.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusRabbitMQ\EventBusRabbitMQ.csproj" />
<ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBusServiceBus\EventBusServiceBus.csproj" />
<ProjectReference Include="..\..\..\BuildingBlocks\EventBus\EventBus\EventBus.csproj" />
<ProjectReference Include="..\..\..\BuildingBlocks\EventBus\IntegrationEventLogEF\IntegrationEventLogEF.csproj" />
<ProjectReference Include="..\..\..\BuildingBlocks\WebHostCustomization\WebHost.Customization\WebHost.Customization.csproj" />
</ItemGroup>