@ -0,0 +1,45 @@ | |||
using Microsoft.eShopOnContainers.Services.Basket.API.Model; | |||
using Microsoft.eShopOnContainers.Services.Common.Infrastructure; | |||
using Microsoft.eShopOnContainers.Services.Common.Infrastructure.Catalog; | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Linq; | |||
using System.Threading.Tasks; | |||
namespace Basket.API.Events | |||
{ | |||
public class ProductPriceChangedHandler : IIntegrationEventHandler<ProductPriceChanged> | |||
{ | |||
private readonly IBasketRepository _repository; | |||
public ProductPriceChangedHandler(IBasketRepository repository) | |||
{ | |||
_repository = repository; | |||
} | |||
public async Task Handle(ProductPriceChanged @event) | |||
{ | |||
var userIds = await _repository.GetUsers(); | |||
foreach (var id in userIds) | |||
{ | |||
var basket = await _repository.GetBasket(id); | |||
await UpdateBasket(@event.ItemId, @event.NewPrice, basket); | |||
} | |||
} | |||
private async Task UpdateBasket(int itemId, decimal newPrice, CustomerBasket basket) | |||
{ | |||
var itemsToUpdate = basket?.Items?.Where(x => int.Parse(x.ProductId) == itemId).ToList(); | |||
if (itemsToUpdate != null) | |||
{ | |||
foreach (var item in itemsToUpdate) | |||
{ | |||
var originalPrice = item.UnitPrice; | |||
item.UnitPrice = newPrice; | |||
item.OldUnitPrice = originalPrice; | |||
} | |||
await _repository.UpdateBasket(basket); | |||
} | |||
} | |||
} | |||
} | |||
@ -0,0 +1,123 @@ | |||
using System; | |||
using Microsoft.EntityFrameworkCore; | |||
using Microsoft.EntityFrameworkCore.Infrastructure; | |||
using Microsoft.EntityFrameworkCore.Metadata; | |||
using Microsoft.EntityFrameworkCore.Migrations; | |||
using Microsoft.eShopOnContainers.Services.Catalog.API.Infrastructure; | |||
using Microsoft.eShopOnContainers.Services.Common.Infrastructure; | |||
namespace Catalog.API.Infrastructure.Migrations | |||
{ | |||
[DbContext(typeof(CatalogContext))] | |||
[Migration("20170314083211_AddEventTable")] | |||
partial class AddEventTable | |||
{ | |||
protected override void BuildTargetModel(ModelBuilder modelBuilder) | |||
{ | |||
modelBuilder | |||
.HasAnnotation("ProductVersion", "1.1.0-rtm-22752") | |||
.HasAnnotation("SqlServer:Sequence:.catalog_brand_hilo", "'catalog_brand_hilo', '', '1', '10', '', '', 'Int64', 'False'") | |||
.HasAnnotation("SqlServer:Sequence:.catalog_hilo", "'catalog_hilo', '', '1', '10', '', '', 'Int64', 'False'") | |||
.HasAnnotation("SqlServer:Sequence:.catalog_type_hilo", "'catalog_type_hilo', '', '1', '10', '', '', 'Int64', 'False'") | |||
.HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.IdentityColumn); | |||
modelBuilder.Entity("Microsoft.eShopOnContainers.Services.Catalog.API.Model.CatalogBrand", b => | |||
{ | |||
b.Property<int>("Id") | |||
.ValueGeneratedOnAdd() | |||
.HasAnnotation("SqlServer:HiLoSequenceName", "catalog_brand_hilo") | |||
.HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.SequenceHiLo); | |||
b.Property<string>("Brand") | |||
.IsRequired() | |||
.HasMaxLength(100); | |||
b.HasKey("Id"); | |||
b.ToTable("CatalogBrand"); | |||
}); | |||
modelBuilder.Entity("Microsoft.eShopOnContainers.Services.Catalog.API.Model.CatalogItem", b => | |||
{ | |||
b.Property<int>("Id") | |||
.ValueGeneratedOnAdd() | |||
.HasAnnotation("SqlServer:HiLoSequenceName", "catalog_hilo") | |||
.HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.SequenceHiLo); | |||
b.Property<int>("CatalogBrandId"); | |||
b.Property<int>("CatalogTypeId"); | |||
b.Property<string>("Description"); | |||
b.Property<string>("Name") | |||
.IsRequired() | |||
.HasMaxLength(50); | |||
b.Property<string>("PictureUri"); | |||
b.Property<decimal>("Price"); | |||
b.HasKey("Id"); | |||
b.HasIndex("CatalogBrandId"); | |||
b.HasIndex("CatalogTypeId"); | |||
b.ToTable("Catalog"); | |||
}); | |||
modelBuilder.Entity("Microsoft.eShopOnContainers.Services.Catalog.API.Model.CatalogType", b => | |||
{ | |||
b.Property<int>("Id") | |||
.ValueGeneratedOnAdd() | |||
.HasAnnotation("SqlServer:HiLoSequenceName", "catalog_type_hilo") | |||
.HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.SequenceHiLo); | |||
b.Property<string>("Type") | |||
.IsRequired() | |||
.HasMaxLength(100); | |||
b.HasKey("Id"); | |||
b.ToTable("CatalogType"); | |||
}); | |||
modelBuilder.Entity("Microsoft.eShopOnContainers.Services.Common.Infrastructure.Data.IntegrationEvent", b => | |||
{ | |||
b.Property<Guid>("EventId") | |||
.ValueGeneratedOnAdd(); | |||
b.Property<string>("Content") | |||
.IsRequired(); | |||
b.Property<DateTime>("CreationTime"); | |||
b.Property<string>("EventTypeName") | |||
.IsRequired() | |||
.HasMaxLength(200); | |||
b.Property<int>("State"); | |||
b.Property<int>("TimesSent"); | |||
b.HasKey("EventId"); | |||
b.ToTable("IntegrationEvent"); | |||
}); | |||
modelBuilder.Entity("Microsoft.eShopOnContainers.Services.Catalog.API.Model.CatalogItem", b => | |||
{ | |||
b.HasOne("Microsoft.eShopOnContainers.Services.Catalog.API.Model.CatalogBrand", "CatalogBrand") | |||
.WithMany() | |||
.HasForeignKey("CatalogBrandId") | |||
.OnDelete(DeleteBehavior.Cascade); | |||
b.HasOne("Microsoft.eShopOnContainers.Services.Catalog.API.Model.CatalogType", "CatalogType") | |||
.WithMany() | |||
.HasForeignKey("CatalogTypeId") | |||
.OnDelete(DeleteBehavior.Cascade); | |||
}); | |||
} | |||
} | |||
} |
@ -0,0 +1,34 @@ | |||
using System; | |||
using System.Collections.Generic; | |||
using Microsoft.EntityFrameworkCore.Migrations; | |||
namespace Catalog.API.Infrastructure.Migrations | |||
{ | |||
public partial class AddEventTable : Migration | |||
{ | |||
protected override void Up(MigrationBuilder migrationBuilder) | |||
{ | |||
migrationBuilder.CreateTable( | |||
name: "IntegrationEvent", | |||
columns: table => new | |||
{ | |||
EventId = table.Column<Guid>(nullable: false), | |||
Content = table.Column<string>(nullable: false), | |||
CreationTime = table.Column<DateTime>(nullable: false), | |||
EventTypeName = table.Column<string>(maxLength: 200, nullable: false), | |||
State = table.Column<int>(nullable: false), | |||
TimesSent = table.Column<int>(nullable: false) | |||
}, | |||
constraints: table => | |||
{ | |||
table.PrimaryKey("PK_IntegrationEvent", x => x.EventId); | |||
}); | |||
} | |||
protected override void Down(MigrationBuilder migrationBuilder) | |||
{ | |||
migrationBuilder.DropTable( | |||
name: "IntegrationEvent"); | |||
} | |||
} | |||
} |
@ -0,0 +1,22 @@ | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Text; | |||
namespace Microsoft.eShopOnContainers.Services.Common.Infrastructure.Catalog | |||
{ | |||
public class ProductPriceChanged : IntegrationEventBase | |||
{ | |||
public int ItemId { get; private set; } | |||
public decimal NewPrice { get; private set; } | |||
public decimal OldPrice { get; set; } | |||
public ProductPriceChanged(int itemId, decimal newPrice, decimal oldPrice) | |||
{ | |||
ItemId = itemId; | |||
NewPrice = newPrice; | |||
OldPrice = oldPrice; | |||
} | |||
} | |||
} |
@ -0,0 +1,13 @@ | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Text; | |||
namespace Microsoft.eShopOnContainers.Services.Common.Infrastructure | |||
{ | |||
public enum EventStateEnum | |||
{ | |||
NotSend = 0, | |||
Sent = 1, | |||
SendingFailed = 2 | |||
} | |||
} |
@ -0,0 +1,26 @@ | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Text; | |||
using Newtonsoft.Json; | |||
namespace Microsoft.eShopOnContainers.Services.Common.Infrastructure.Data | |||
{ | |||
public class IntegrationEvent | |||
{ | |||
public IntegrationEvent(IntegrationEventBase @event) | |||
{ | |||
EventId = @event.Id; | |||
CreationTime = DateTime.UtcNow; | |||
EventTypeName = @event.GetType().FullName; | |||
Content = JsonConvert.SerializeObject(@event); | |||
State = EventStateEnum.NotSend; | |||
TimesSent = 0; | |||
} | |||
public Guid EventId { get; private set; } | |||
public string EventTypeName { get; private set; } | |||
public EventStateEnum State { get; set; } | |||
public int TimesSent { get; set; } | |||
public DateTime CreationTime { get; private set; } | |||
public string Content { get; private set; } | |||
} | |||
} |
@ -0,0 +1,154 @@ | |||
| |||
using Microsoft.eShopOnContainers.Services.Common.Infrastructure.Catalog; | |||
using Newtonsoft.Json; | |||
using RabbitMQ.Client; | |||
using RabbitMQ.Client.Events; | |||
using System; | |||
using System.Collections; | |||
using System.Collections.Generic; | |||
using System.Linq; | |||
using System.Reflection; | |||
using System.Text; | |||
using System.Threading.Tasks; | |||
namespace Microsoft.eShopOnContainers.Services.Common.Infrastructure | |||
{ | |||
public class EventBusRabbitMQ : IEventBus | |||
{ | |||
private readonly string _brokerName = "eshop_event_bus"; | |||
private readonly string _connectionString; | |||
private readonly Dictionary<string, List<IIntegrationEventHandler>> _handlers; | |||
private readonly List<Type> _eventTypes; | |||
private Tuple<IModel, IConnection> _connection; | |||
private string _queueName; | |||
public EventBusRabbitMQ(string connectionString) | |||
{ | |||
_connectionString = connectionString; | |||
_handlers = new Dictionary<string, List<IIntegrationEventHandler>>(); | |||
_eventTypes = new List<Type>(); | |||
} | |||
public void Publish(IntegrationEventBase @event) | |||
{ | |||
var eventName = @event.GetType().Name; | |||
var factory = new ConnectionFactory() { HostName = _connectionString }; | |||
using (var connection = factory.CreateConnection()) | |||
using (var channel = connection.CreateModel()) | |||
{ | |||
channel.ExchangeDeclare(exchange: _brokerName, | |||
type: "direct"); | |||
string message = JsonConvert.SerializeObject(@event); | |||
var body = Encoding.UTF8.GetBytes(message); | |||
channel.BasicPublish(exchange: _brokerName, | |||
routingKey: eventName, | |||
basicProperties: null, | |||
body: body); | |||
} | |||
} | |||
public void Subscribe<T>(IIntegrationEventHandler<T> handler) where T : IntegrationEventBase | |||
{ | |||
var eventName = typeof(T).Name; | |||
if (_handlers.ContainsKey(eventName)) | |||
{ | |||
_handlers[eventName].Add(handler); | |||
} | |||
else | |||
{ | |||
var channel = GetChannel(); | |||
channel.QueueBind(queue: _queueName, | |||
exchange: _brokerName, | |||
routingKey: eventName); | |||
_handlers.Add(eventName, new List<IIntegrationEventHandler>()); | |||
_handlers[eventName].Add(handler); | |||
_eventTypes.Add(typeof(T)); | |||
} | |||
} | |||
public void Unsubscribe<T>(IIntegrationEventHandler<T> handler) where T : IntegrationEventBase | |||
{ | |||
var eventName = typeof(T).Name; | |||
if (_handlers.ContainsKey(eventName) && _handlers[eventName].Contains(handler)) | |||
{ | |||
_handlers[eventName].Remove(handler); | |||
if (_handlers[eventName].Count == 0) | |||
{ | |||
_handlers.Remove(eventName); | |||
var eventType = _eventTypes.Single(e => e.Name == eventName); | |||
_eventTypes.Remove(eventType); | |||
_connection.Item1.QueueUnbind(queue: _queueName, | |||
exchange: _brokerName, | |||
routingKey: eventName); | |||
if (_handlers.Keys.Count == 0) | |||
{ | |||
_queueName = string.Empty; | |||
_connection.Item1.Close(); | |||
_connection.Item2.Close(); | |||
} | |||
} | |||
} | |||
} | |||
private IModel GetChannel() | |||
{ | |||
if (_connection != null) | |||
{ | |||
return _connection.Item1; | |||
} | |||
else | |||
{ | |||
var factory = new ConnectionFactory() { HostName = _connectionString }; | |||
var connection = factory.CreateConnection(); | |||
var channel = connection.CreateModel(); | |||
channel.ExchangeDeclare(exchange: _brokerName, | |||
type: "direct"); | |||
if (string.IsNullOrEmpty(_queueName)) | |||
{ | |||
_queueName = channel.QueueDeclare().QueueName; | |||
} | |||
var consumer = new EventingBasicConsumer(channel); | |||
consumer.Received += async (model, ea) => | |||
{ | |||
var eventName = ea.RoutingKey; | |||
var message = Encoding.UTF8.GetString(ea.Body); | |||
await ProcessEvent(eventName, message); | |||
}; | |||
channel.BasicConsume(queue: _queueName, | |||
noAck: true, | |||
consumer: consumer); | |||
_connection = new Tuple<IModel, IConnection>(channel, connection); | |||
return _connection.Item1; | |||
} | |||
} | |||
private async Task ProcessEvent(string eventName, string message) | |||
{ | |||
if (_handlers.ContainsKey(eventName)) | |||
{ | |||
Type eventType = _eventTypes.Single(t => t.Name == eventName); | |||
var integrationEvent = JsonConvert.DeserializeObject(message, eventType); | |||
var concreteType = typeof(IIntegrationEventHandler<>).MakeGenericType(eventType); | |||
var handlers = _handlers[eventName]; | |||
foreach (var handler in handlers) | |||
{ | |||
await (Task)concreteType.GetMethod("Handle").Invoke(handler, new object[] { integrationEvent }); | |||
} | |||
} | |||
} | |||
} | |||
} |
@ -0,0 +1,13 @@ | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Text; | |||
namespace Microsoft.eShopOnContainers.Services.Common.Infrastructure | |||
{ | |||
public interface IEventBus | |||
{ | |||
void Subscribe<T>(IIntegrationEventHandler<T> handler) where T: IntegrationEventBase; | |||
void Unsubscribe<T>(IIntegrationEventHandler<T> handler) where T : IntegrationEventBase; | |||
void Publish(IntegrationEventBase @event); | |||
} | |||
} |
@ -0,0 +1,17 @@ | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Text; | |||
using System.Threading.Tasks; | |||
namespace Microsoft.eShopOnContainers.Services.Common.Infrastructure | |||
{ | |||
public interface IIntegrationEventHandler<in TIntegrationEvent> : IIntegrationEventHandler | |||
where TIntegrationEvent: IntegrationEventBase | |||
{ | |||
Task Handle(TIntegrationEvent @event); | |||
} | |||
public interface IIntegrationEventHandler | |||
{ | |||
} | |||
} |
@ -0,0 +1,11 @@ | |||
<Project Sdk="Microsoft.NET.Sdk" ToolsVersion="15.0"> | |||
<PropertyGroup> | |||
<TargetFramework>netcoreapp1.1</TargetFramework> | |||
<RuntimeFrameworkVersion>1.1.0</RuntimeFrameworkVersion> | |||
<RootNamespace>Microsoft.eShopOnContainers.Services.Common.Infrastructure</RootNamespace> | |||
</PropertyGroup> | |||
<ItemGroup> | |||
<PackageReference Include="Newtonsoft.Json" Version="9.0.1" /> | |||
<PackageReference Include="RabbitMQ.Client" Version="4.1.1" /> | |||
</ItemGroup> | |||
</Project> |
@ -0,0 +1,16 @@ | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Text; | |||
namespace Microsoft.eShopOnContainers.Services.Common.Infrastructure | |||
{ | |||
public class IntegrationEventBase | |||
{ | |||
public IntegrationEventBase() | |||
{ | |||
Id = Guid.NewGuid(); | |||
} | |||
public Guid Id { get; private set; } | |||
} | |||
} |
@ -0,0 +1,55 @@ | |||
using Microsoft.eShopOnContainers.Services.Catalog.API.Model; | |||
using Microsoft.eShopOnContainers.Services.Catalog.API.ViewModel; | |||
using Newtonsoft.Json; | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Linq; | |||
using System.Net.Http; | |||
using System.Text; | |||
using System.Threading.Tasks; | |||
using Xunit; | |||
namespace FunctionalTests.Services.Catalog | |||
{ | |||
public class CatalogScenarios : CatalogScenariosBase | |||
{ | |||
[Fact] | |||
public async Task Post_update_a_catalogitem_price_and_catalogitem_is_returned_modified() | |||
{ | |||
using (var server = CreateServer()) | |||
{ | |||
var client = server.CreateClient(); | |||
// Arrange | |||
var itemToModify = GetCatalogItem(); | |||
var newPrice = new Random().Next(1, 200); | |||
itemToModify.Price = newPrice; | |||
// Act | |||
var postRes = await client.PostAsync(Post.UpdateCatalogProduct, | |||
new StringContent(JsonConvert.SerializeObject(itemToModify), | |||
UTF8Encoding.UTF8, "application/json")); | |||
var response = await client.GetAsync(Get.ProductByName(itemToModify.Name)); | |||
var result = JsonConvert.DeserializeObject<PaginatedItemsViewModel<CatalogItem>>(await response.Content.ReadAsStringAsync()); | |||
var item = result.Data.First(); | |||
// Assert | |||
Assert.Equal(result.Count, 1); | |||
Assert.Equal(itemToModify.Id, item.Id); | |||
Assert.Equal(newPrice, item.Price); | |||
} | |||
} | |||
private CatalogItem GetCatalogItem() | |||
{ | |||
return new CatalogItem() | |||
{ | |||
Id = 1, | |||
Price = 12.5M, | |||
Name = ".NET Bot Black Sweatshirt" | |||
}; | |||
} | |||
} | |||
} |
@ -0,0 +1,38 @@ | |||
using FunctionalTests.Middleware; | |||
using Microsoft.AspNetCore.Hosting; | |||
using Microsoft.AspNetCore.TestHost; | |||
using Microsoft.eShopOnContainers.Services.Catalog.API; | |||
using System; | |||
using System.Collections.Generic; | |||
using System.IO; | |||
using System.Text; | |||
namespace FunctionalTests.Services.Catalog | |||
{ | |||
public class CatalogScenariosBase | |||
{ | |||
public TestServer CreateServer() | |||
{ | |||
var webHostBuilder = new WebHostBuilder(); | |||
webHostBuilder.UseContentRoot(Directory.GetCurrentDirectory()); | |||
webHostBuilder.UseStartup<Startup>(); | |||
return new TestServer(webHostBuilder); | |||
} | |||
public static class Get | |||
{ | |||
public static string Orders = "api/v1/orders"; | |||
public static string ProductByName(string name) | |||
{ | |||
return $"api/v1/catalog/items/withname/{name}"; | |||
} | |||
} | |||
public static class Post | |||
{ | |||
public static string UpdateCatalogProduct = "api/v1/catalog"; | |||
} | |||
} | |||
} |
@ -0,0 +1,15 @@ | |||
using Microsoft.eShopOnContainers.Services.Catalog.API; | |||
using System; | |||
using System.Collections.Generic; | |||
using System.Text; | |||
using Microsoft.AspNetCore.Hosting; | |||
namespace FunctionalTests.Services.Catalog | |||
{ | |||
public class CatalogTestsStartup : Startup | |||
{ | |||
public CatalogTestsStartup(IHostingEnvironment env) : base(env) | |||
{ | |||
} | |||
} | |||
} |
@ -1,5 +1,6 @@ | |||
{ | |||
"ConnectionString": "Server=tcp:127.0.0.1,5433;Database=Microsoft.eShopOnContainers.Services.OrderingDb;User Id=sa;Password=Pass@word;", | |||
"IdentityUrl": "http://localhost:5105", | |||
"isTest": "true" | |||
"isTest": "true", | |||
"EventBusConnection": "localhost" | |||
} |