using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.eShopOnContainers.BuildingBlocks.EventBus; using Microsoft.eShopOnContainers.BuildingBlocks.EventBus.Events; using Newtonsoft.Json; using System; using System.Collections; using System.Collections.Generic; using System.Data.Common; using System.Linq; using System.Threading.Tasks; namespace Microsoft.eShopOnContainers.BuildingBlocks.IntegrationEventLogEF.Services { public class IntegrationEventLogService : IIntegrationEventLogService { private readonly IEventBusSubscriptionsManager _subsManager; private readonly IntegrationEventLogContext _integrationEventLogContext; private readonly DbConnection _dbConnection; public IntegrationEventLogService(IEventBusSubscriptionsManager subsManager, DbConnection dbConnection) { _dbConnection = dbConnection ?? throw new ArgumentNullException(nameof(dbConnection)); _subsManager = subsManager ?? throw new ArgumentNullException(nameof(subsManager)); _integrationEventLogContext = new IntegrationEventLogContext( new DbContextOptionsBuilder() .UseSqlServer(_dbConnection) .ConfigureWarnings(warnings => warnings.Throw(RelationalEventId.QueryClientEvaluationWarning)) .Options); } public async Task> RetrieveEventLogsPendingToPublishAsync() { var eventLogsPendingToPublish = await _integrationEventLogContext.IntegrationEventLogs .Where(e => e.State == EventStateEnum.NotPublished) .OrderBy(o => o.CreationTime) .ToListAsync(); eventLogsPendingToPublish.ForEach(evtLog => evtLog.DeserializeJsonContent(_subsManager.GetEventTypeByName(evtLog.EventTypeShortName))); return eventLogsPendingToPublish; } public Task SaveEventAsync(IntegrationEvent @event, DbTransaction transaction) { if (transaction == null) { throw new ArgumentNullException(nameof(transaction), $"A {typeof(DbTransaction).FullName} is required as a pre-requisite to save the event."); } var eventLogEntry = new IntegrationEventLogEntry(@event); _integrationEventLogContext.Database.UseTransaction(transaction); _integrationEventLogContext.IntegrationEventLogs.Add(eventLogEntry); return _integrationEventLogContext.SaveChangesAsync(); } public Task MarkEventAsPublishedAsync(Guid eventId) { return UpdateEventStatus(eventId, EventStateEnum.Published); } public Task MarkEventAsInProgressAsync(Guid eventId) { return UpdateEventStatus(eventId, EventStateEnum.InProgress); } public Task MarkEventAsFailedAsync(Guid eventId) { return UpdateEventStatus(eventId, EventStateEnum.PublishedFailed); } private Task UpdateEventStatus(Guid eventId, EventStateEnum status) { var eventLogEntry = _integrationEventLogContext.IntegrationEventLogs.Single(ie => ie.EventId == eventId); eventLogEntry.State = status; if(status == EventStateEnum.InProgress) eventLogEntry.TimesSent++; _integrationEventLogContext.IntegrationEventLogs.Update(eventLogEntry); return _integrationEventLogContext.SaveChangesAsync(); } } }