diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs index 2b0a448f74..86fc78cab3 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs +++ b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs @@ -1,26 +1,31 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Options; using Volo.Abp.EventBus.Distributed; using Volo.Abp.Timing; using Volo.Abp.Uow; namespace Volo.Abp.EntityFrameworkCore.DistributedEvents { - public class DbContextEventInbox : IDbContextEventInbox + public class DbContextEventInbox : IDbContextEventInbox where TDbContext : IHasEventInbox { protected IDbContextProvider DbContextProvider { get; } + protected AbpDistributedEventBusOptions DistributedEventsOptions { get; } protected IClock Clock { get; } public DbContextEventInbox( IDbContextProvider dbContextProvider, - IClock clock) + IClock clock, + IOptions distributedEventsOptions) { DbContextProvider = dbContextProvider; Clock = clock; + DistributedEventsOptions = distributedEventsOptions.Value; } [UnitOfWork] @@ -34,7 +39,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents } [UnitOfWork] - public virtual async Task> GetWaitingEventsAsync(int maxCount) + public virtual async Task> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default) { var dbContext = await DbContextProvider.GetDbContextAsync(); @@ -44,8 +49,8 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents .Where(x => !x.Processed) .OrderBy(x => x.CreationTime) .Take(maxCount) - .ToListAsync(); - + .ToListAsync(cancellationToken: cancellationToken); + return outgoingEventRecords .Select(x => x.ToIncomingEventInfo()) .ToList(); @@ -76,11 +81,11 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents { //TODO: Optimize var dbContext = await DbContextProvider.GetDbContextAsync(); - var timeToKeepEvents = Clock.Now.AddHours(-2); //TODO: Config? + var timeToKeepEvents = Clock.Now.Add(DistributedEventsOptions.InboxKeepEventTimeSpan); var oldEvents = await dbContext.IncomingEvents .Where(x => x.Processed && x.CreationTime < timeToKeepEvents) .ToListAsync(); dbContext.IncomingEvents.RemoveRange(oldEvents); } } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs index a785ad7ce5..435bc70398 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs +++ b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; using Volo.Abp.EventBus.Distributed; @@ -8,7 +9,7 @@ using Volo.Abp.Uow; namespace Volo.Abp.EntityFrameworkCore.DistributedEvents { - public class DbContextEventOutbox : IDbContextEventOutbox + public class DbContextEventOutbox : IDbContextEventOutbox where TDbContext : IHasEventOutbox { protected IDbContextProvider DbContextProvider { get; } @@ -18,7 +19,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents { DbContextProvider = dbContextProvider; } - + [UnitOfWork] public virtual async Task EnqueueAsync(OutgoingEventInfo outgoingEvent) { @@ -29,17 +30,17 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents } [UnitOfWork] - public virtual async Task> GetWaitingEventsAsync(int maxCount) + public virtual async Task> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default) { var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); - + var outgoingEventRecords = await dbContext .OutgoingEvents .AsNoTracking() .OrderBy(x => x.CreationTime) .Take(maxCount) - .ToListAsync(); - + .ToListAsync(cancellationToken: cancellationToken); + return outgoingEventRecords .Select(x => x.ToOutgoingEventInfo()) .ToList(); @@ -57,4 +58,4 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents } } } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpEventBusBoxesOptions.cs b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpEventBusBoxesOptions.cs new file mode 100644 index 0000000000..da91d672d3 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpEventBusBoxesOptions.cs @@ -0,0 +1,43 @@ +using System; + +namespace Volo.Abp.EventBus.Boxes +{ + public class AbpEventBusBoxesOptions + { + /// + /// Default: 6 hours + /// + public TimeSpan CleanOldEventTimeIntervalSpan { get; set; } + + /// + /// Default: 1000 + /// + public int InboxWaitingEventMaxCount { get; set; } + + /// + /// Default: 1000 + /// + public int OutboxWaitingEventMaxCount { get; set; } + + /// + /// Period time of and + /// Default: 2 seconds + /// + public TimeSpan PeriodTimeSpan { get; set; } + + /// + /// Delay time of and + /// Default: 15 seconds + /// + public TimeSpan DelayTimeSpan { get; set; } + + public AbpEventBusBoxesOptions() + { + CleanOldEventTimeIntervalSpan = TimeSpan.FromHours(6); + InboxWaitingEventMaxCount = 1000; + OutboxWaitingEventMaxCount = 1000; + PeriodTimeSpan = TimeSpan.FromSeconds(2); + DelayTimeSpan = TimeSpan.FromSeconds(15); + } + } +} diff --git a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs index e711e5798e..29cca3d624 100644 --- a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs +++ b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs @@ -5,6 +5,7 @@ using Medallion.Threading; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; using Volo.Abp.DependencyInjection; using Volo.Abp.EventBus.Distributed; using Volo.Abp.Threading; @@ -23,6 +24,7 @@ namespace Volo.Abp.EventBus.Boxes protected IClock Clock { get; } protected IEventInbox Inbox { get; private set; } protected InboxConfig InboxConfig { get; private set; } + protected AbpEventBusBoxesOptions EventBusBoxesOptions { get; } protected DateTime? LastCleanTime { get; set; } @@ -37,7 +39,8 @@ namespace Volo.Abp.EventBus.Boxes IDistributedEventBus distributedEventBus, IDistributedLockProvider distributedLockProvider, IUnitOfWorkManager unitOfWorkManager, - IClock clock) + IClock clock, + IOptions eventBusBoxesOptions) { ServiceProvider = serviceProvider; Timer = timer; @@ -45,7 +48,8 @@ namespace Volo.Abp.EventBus.Boxes DistributedLockProvider = distributedLockProvider; UnitOfWorkManager = unitOfWorkManager; Clock = clock; - Timer.Period = 2000; //TODO: Config? + EventBusBoxesOptions = eventBusBoxesOptions.Value; + Timer.Period = EventBusBoxesOptions.PeriodTimeSpan.Seconds; Timer.Elapsed += TimerOnElapsed; Logger = NullLogger.Instance; StoppingTokenSource = new CancellationTokenSource(); @@ -79,7 +83,7 @@ namespace Volo.Abp.EventBus.Boxes { return; } - + await using (var handle = await DistributedLockProvider.TryAcquireLockAsync(DistributedLockName, cancellationToken: StoppingToken)) { if (handle != null) @@ -88,7 +92,7 @@ namespace Volo.Abp.EventBus.Boxes while (true) { - var waitingEvents = await Inbox.GetWaitingEventsAsync(1000); //TODO: Config? Pass StoppingToken! + var waitingEvents = await Inbox.GetWaitingEventsAsync(EventBusBoxesOptions.InboxWaitingEventMaxCount, StoppingToken); if (waitingEvents.Count <= 0) { break; @@ -116,14 +120,14 @@ namespace Volo.Abp.EventBus.Boxes else { Logger.LogDebug("Could not obtain the distributed lock: " + DistributedLockName); - await TaskDelayHelper.DelayAsync(15000, StoppingToken); //TODO: Config? + await TaskDelayHelper.DelayAsync(EventBusBoxesOptions.DelayTimeSpan.Milliseconds, StoppingToken); } } } protected virtual async Task DeleteOldEventsAsync() { - if (LastCleanTime != null && LastCleanTime > Clock.Now.AddHours(6)) //TODO: Config? + if (LastCleanTime != null && LastCleanTime > Clock.Now.Add(EventBusBoxesOptions.CleanOldEventTimeIntervalSpan)) { return; } diff --git a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs index 1be2374268..e8a05c60ba 100644 --- a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs +++ b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs @@ -5,6 +5,7 @@ using Medallion.Threading; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; using Volo.Abp.DependencyInjection; using Volo.Abp.EventBus.Distributed; using Volo.Abp.Threading; @@ -19,9 +20,10 @@ namespace Volo.Abp.EventBus.Boxes protected IDistributedLockProvider DistributedLockProvider { get; } protected IEventOutbox Outbox { get; private set; } protected OutboxConfig OutboxConfig { get; private set; } + protected AbpEventBusBoxesOptions EventBusBoxesOptions { get; } protected string DistributedLockName => "Outbox_" + OutboxConfig.Name; public ILogger Logger { get; set; } - + protected CancellationTokenSource StoppingTokenSource { get; } protected CancellationToken StoppingToken { get; } @@ -29,13 +31,15 @@ namespace Volo.Abp.EventBus.Boxes IServiceProvider serviceProvider, AbpAsyncTimer timer, IDistributedEventBus distributedEventBus, - IDistributedLockProvider distributedLockProvider) + IDistributedLockProvider distributedLockProvider, + IOptions eventBusBoxesOptions) { ServiceProvider = serviceProvider; Timer = timer; DistributedEventBus = distributedEventBus; DistributedLockProvider = distributedLockProvider; - Timer.Period = 2000; //TODO: Config? + EventBusBoxesOptions = eventBusBoxesOptions.Value; + Timer.Period = EventBusBoxesOptions.PeriodTimeSpan.Seconds; Timer.Elapsed += TimerOnElapsed; Logger = NullLogger.Instance; StoppingTokenSource = new CancellationTokenSource(); @@ -65,13 +69,13 @@ namespace Volo.Abp.EventBus.Boxes protected virtual async Task RunAsync() { - await using (var handle = await DistributedLockProvider.TryAcquireLockAsync(DistributedLockName)) + await using (var handle = await DistributedLockProvider.TryAcquireLockAsync(DistributedLockName, cancellationToken: StoppingToken)) { if (handle != null) { while (true) { - var waitingEvents = await Outbox.GetWaitingEventsAsync(1000); //TODO: Config? + var waitingEvents = await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, StoppingToken); if (waitingEvents.Count <= 0) { break; @@ -96,7 +100,7 @@ namespace Volo.Abp.EventBus.Boxes else { Logger.LogDebug("Could not obtain the distributed lock: " + DistributedLockName); - await TaskDelayHelper.DelayAsync(15000, StoppingToken); //TODO: Config? + await TaskDelayHelper.DelayAsync(EventBusBoxesOptions.DelayTimeSpan.Milliseconds, StoppingToken); } } } diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/IRabbitMqSerializer.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/IRebusSerializer.cs similarity index 100% rename from framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/IRabbitMqSerializer.cs rename to framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/IRebusSerializer.cs diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs index 8dac885036..1cb380679f 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs @@ -6,6 +6,7 @@ using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using Rebus.Bus; +using Rebus.Pipeline; using Volo.Abp.DependencyInjection; using Volo.Abp.EventBus.Distributed; using Volo.Abp.Guids; @@ -134,6 +135,19 @@ namespace Volo.Abp.EventBus.Rebus Rebus.Unsubscribe(eventType); } + public async Task ProcessEventAsync(Type eventType, object eventData) + { + var messageId = MessageContext.Current.TransportMessage.GetMessageId(); + var eventName = EventNameAttribute.GetNameOrDefault(eventType); + + if (await AddToInboxAsync(messageId, eventName, eventType, MessageContext.Current.TransportMessage.Body)) + { + return; + } + + await TriggerHandlersAsync(eventType, eventData); + } + protected override async Task PublishToEventBusAsync(Type eventType, object eventData) { await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData); @@ -192,16 +206,29 @@ namespace Volo.Abp.EventBus.Rebus OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) { - /* TODO: IMPLEMENT! */ - throw new NotImplementedException(); + var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName); + var eventData = Serializer.Deserialize(outgoingEvent.EventData, eventType); + + return PublishToEventBusAsync(eventType, eventData); } - public override Task ProcessFromInboxAsync( + public override async Task ProcessFromInboxAsync( IncomingEventInfo incomingEvent, InboxConfig inboxConfig) { - /* TODO: IMPLEMENT! */ - throw new NotImplementedException(); + var eventType = EventTypes.GetOrDefault(incomingEvent.EventName); + if (eventType == null) + { + return; + } + + var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType); + var exceptions = new List(); + await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); + if (exceptions.Any()) + { + ThrowOriginalExceptions(eventType, exceptions); + } } protected override byte[] Serialize(object eventData) diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventHandlerAdapter.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventHandlerAdapter.cs index 005226b885..61330828b4 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventHandlerAdapter.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventHandlerAdapter.cs @@ -14,7 +14,7 @@ namespace Volo.Abp.EventBus.Rebus public async Task Handle(TEventData message) { - await RebusDistributedEventBus.TriggerHandlersAsync(typeof(TEventData), message); + await RebusDistributedEventBus.ProcessEventAsync(message.GetType(), message); } } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpDistributedEventBusOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpDistributedEventBusOptions.cs index 896dfd4f7d..5b71542665 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpDistributedEventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpDistributedEventBusOptions.cs @@ -1,3 +1,4 @@ +using System; using Volo.Abp.Collections; namespace Volo.Abp.EventBus.Distributed @@ -5,16 +6,22 @@ namespace Volo.Abp.EventBus.Distributed public class AbpDistributedEventBusOptions { public ITypeList Handlers { get; } - + public OutboxConfigDictionary Outboxes { get; } - + public InboxConfigDictionary Inboxes { get; } + /// + /// Default: -2 hours + /// + public TimeSpan InboxKeepEventTimeSpan { get; set; } + public AbpDistributedEventBusOptions() { Handlers = new TypeList(); Outboxes = new OutboxConfigDictionary(); Inboxes = new InboxConfigDictionary(); + InboxKeepEventTimeSpan = TimeSpan.FromHours(-2); } } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs index bee802a126..d072f5d922 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; namespace Volo.Abp.EventBus.Distributed @@ -7,13 +8,13 @@ namespace Volo.Abp.EventBus.Distributed public interface IEventInbox { Task EnqueueAsync(IncomingEventInfo incomingEvent); - - Task> GetWaitingEventsAsync(int maxCount); - + + Task> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default); + Task MarkAsProcessedAsync(Guid id); - + Task ExistsByMessageIdAsync(string messageId); Task DeleteOldEventsAsync(); } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs index 0b50993e8e..cc00ee6654 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; namespace Volo.Abp.EventBus.Distributed @@ -7,9 +8,9 @@ namespace Volo.Abp.EventBus.Distributed public interface IEventOutbox { Task EnqueueAsync(OutgoingEventInfo outgoingEvent); - - Task> GetWaitingEventsAsync(int maxCount); - + + Task> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default); + Task DeleteAsync(Guid id); } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs index d98e144ada..886a7744d1 100644 --- a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs +++ b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs @@ -1,7 +1,9 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; +using Microsoft.Extensions.Options; using MongoDB.Driver; using MongoDB.Driver.Linq; using Volo.Abp.EventBus.Distributed; @@ -10,21 +12,24 @@ using Volo.Abp.Uow; namespace Volo.Abp.MongoDB.DistributedEvents { - public class MongoDbContextEventInbox : IMongoDbContextEventInbox + public class MongoDbContextEventInbox : IMongoDbContextEventInbox where TMongoDbContext : IHasEventInbox { protected IMongoDbContextProvider DbContextProvider { get; } + protected AbpDistributedEventBusOptions DistributedEventsOptions { get; } protected IClock Clock { get; } - + public MongoDbContextEventInbox( IMongoDbContextProvider dbContextProvider, - IClock clock) + IClock clock, + IOptions distributedEventsOptions) { DbContextProvider = dbContextProvider; Clock = clock; + DistributedEventsOptions = distributedEventsOptions.Value; } - + [UnitOfWork] public virtual async Task EnqueueAsync(IncomingEventInfo incomingEvent) { @@ -45,9 +50,9 @@ namespace Volo.Abp.MongoDB.DistributedEvents } [UnitOfWork] - public virtual async Task> GetWaitingEventsAsync(int maxCount) + public virtual async Task> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default) { - var dbContext = await DbContextProvider.GetDbContextAsync(); + var dbContext = await DbContextProvider.GetDbContextAsync(cancellationToken); var outgoingEventRecords = await dbContext .IncomingEvents @@ -55,8 +60,8 @@ namespace Volo.Abp.MongoDB.DistributedEvents .Where(x => !x.Processed) .OrderBy(x => x.CreationTime) .Take(maxCount) - .ToListAsync(); - + .ToListAsync(cancellationToken: cancellationToken); + return outgoingEventRecords .Select(x => x.ToIncomingEventInfo()) .ToList(); @@ -65,17 +70,44 @@ namespace Volo.Abp.MongoDB.DistributedEvents [UnitOfWork] public async Task MarkAsProcessedAsync(Guid id) { - throw new NotImplementedException(); + var dbContext = await DbContextProvider.GetDbContextAsync(); + var incomingEvent = await dbContext.IncomingEvents.Find(x => x.Id.Equals(id)).FirstOrDefaultAsync(); + if (incomingEvent != null) + { + incomingEvent.MarkAsProcessed(Clock.Now); + + if (dbContext.SessionHandle != null) + { + await dbContext.IncomingEvents.ReplaceOneAsync(dbContext.SessionHandle, Builders.Filter.Eq(e => e.Id, incomingEvent.Id), incomingEvent); + } + else + { + await dbContext.IncomingEvents.ReplaceOneAsync(Builders.Filter.Eq(e => e.Id, incomingEvent.Id), incomingEvent); + } + } } - public Task ExistsByMessageIdAsync(string messageId) + [UnitOfWork] + public async Task ExistsByMessageIdAsync(string messageId) { - throw new NotImplementedException(); + var dbContext = await DbContextProvider.GetDbContextAsync(); + return await dbContext.IncomingEvents.AsQueryable().AnyAsync(x => x.MessageId == messageId); } - public Task DeleteOldEventsAsync() + [UnitOfWork] + public async Task DeleteOldEventsAsync() { - throw new NotImplementedException(); + var dbContext = await DbContextProvider.GetDbContextAsync(); + var timeToKeepEvents = Clock.Now.Add(DistributedEventsOptions.InboxKeepEventTimeSpan); + + if (dbContext.SessionHandle != null) + { + await dbContext.IncomingEvents.DeleteManyAsync(dbContext.SessionHandle, x => x.Processed && x.CreationTime < timeToKeepEvents); + } + else + { + await dbContext.IncomingEvents.DeleteManyAsync(x => x.Processed && x.CreationTime < timeToKeepEvents); + } } } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs index 1bec9e0834..df8aa108ee 100644 --- a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs +++ b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs @@ -1,27 +1,72 @@ using System; using System.Collections.Generic; +using System.Linq; +using System.Threading; using System.Threading.Tasks; +using MongoDB.Driver; +using MongoDB.Driver.Linq; using Volo.Abp.EventBus.Distributed; using Volo.Abp.Uow; namespace Volo.Abp.MongoDB.DistributedEvents { - public class MongoDbContextEventOutbox : IMongoDbContextEventOutbox + public class MongoDbContextEventOutbox : IMongoDbContextEventOutbox where TMongoDbContext : IHasEventOutbox { - public Task EnqueueAsync(OutgoingEventInfo outgoingEvent) + protected IMongoDbContextProvider MongoDbContextProvider { get; } + + public MongoDbContextEventOutbox(IMongoDbContextProvider mongoDbContextProvider) + { + MongoDbContextProvider = mongoDbContextProvider; + } + + [UnitOfWork] + public async Task EnqueueAsync(OutgoingEventInfo outgoingEvent) { - throw new NotImplementedException(); + var dbContext = (IHasEventOutbox) await MongoDbContextProvider.GetDbContextAsync(); + if (dbContext.SessionHandle != null) + { + await dbContext.OutgoingEvents.InsertOneAsync( + dbContext.SessionHandle, + new OutgoingEventRecord(outgoingEvent) + ); + } + else + { + await dbContext.OutgoingEvents.InsertOneAsync( + new OutgoingEventRecord(outgoingEvent) + ); + } } - public Task> GetWaitingEventsAsync(int maxCount) + [UnitOfWork] + public async Task> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default) { - throw new NotImplementedException(); + var dbContext = (IHasEventOutbox) await MongoDbContextProvider.GetDbContextAsync(cancellationToken); + + var outgoingEventRecords = await dbContext + .OutgoingEvents.AsQueryable() + .OrderBy(x => x.CreationTime) + .Take(maxCount) + .ToListAsync(cancellationToken: cancellationToken); + + return outgoingEventRecords + .Select(x => x.ToOutgoingEventInfo()) + .ToList(); } - public Task DeleteAsync(Guid id) + [UnitOfWork] + public async Task DeleteAsync(Guid id) { - throw new NotImplementedException(); + var dbContext = (IHasEventOutbox) await MongoDbContextProvider.GetDbContextAsync(); + if (dbContext.SessionHandle != null) + { + await dbContext.OutgoingEvents.DeleteOneAsync(dbContext.SessionHandle, x => x.Id.Equals(id)); + } + else + { + await dbContext.OutgoingEvents.DeleteOneAsync(x => x.Id.Equals(id)); + } } } -} \ No newline at end of file +} diff --git a/test/DistEvents/DistDemoApp.EfCoreRabbitMq/appsettings.json b/test/DistEvents/DistDemoApp.EfCoreRabbitMq/appsettings.json index 1ed30c80a9..5359c021b7 100644 --- a/test/DistEvents/DistDemoApp.EfCoreRabbitMq/appsettings.json +++ b/test/DistEvents/DistDemoApp.EfCoreRabbitMq/appsettings.json @@ -16,4 +16,4 @@ "Redis": { "Configuration": "127.0.0.1" } -} \ No newline at end of file +} diff --git a/test/DistEvents/DistDemoApp.MongoDbKafka/DistDemoApp.MongoDbKafka.csproj b/test/DistEvents/DistDemoApp.MongoDbKafka/DistDemoApp.MongoDbKafka.csproj index cfd73d51bb..a503c6f6c0 100644 --- a/test/DistEvents/DistDemoApp.MongoDbKafka/DistDemoApp.MongoDbKafka.csproj +++ b/test/DistEvents/DistDemoApp.MongoDbKafka/DistDemoApp.MongoDbKafka.csproj @@ -6,4 +6,16 @@ DistDemoApp + + + + + + + + + Always + + + diff --git a/test/DistEvents/DistDemoApp.MongoDbKafka/DistDemoAppMongoDbKafkaModule.cs b/test/DistEvents/DistDemoApp.MongoDbKafka/DistDemoAppMongoDbKafkaModule.cs new file mode 100644 index 0000000000..b2e41b6ca7 --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbKafka/DistDemoAppMongoDbKafkaModule.cs @@ -0,0 +1,38 @@ +using Microsoft.Extensions.DependencyInjection; +using Volo.Abp.EventBus.Distributed; +using Volo.Abp.EventBus.Kafka; +using Volo.Abp.Modularity; +using Volo.Abp.MongoDB; +using Volo.Abp.MongoDB.DistributedEvents; + +namespace DistDemoApp +{ + [DependsOn( + typeof(AbpMongoDbModule), + typeof(AbpEventBusKafkaModule), + typeof(DistDemoAppSharedModule) + )] + public class DistDemoAppMongoDbKafkaModule : AbpModule + { + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddMongoDbContext(options => + { + options.AddDefaultRepositories(); + }); + + Configure(options => + { + options.Outboxes.Configure(config => + { + config.UseMongoDbContext(); + }); + + options.Inboxes.Configure(config => + { + config.UseMongoDbContext(); + }); + }); + } + } +} diff --git a/test/DistEvents/DistDemoApp.MongoDbKafka/Program.cs b/test/DistEvents/DistDemoApp.MongoDbKafka/Program.cs index d0e4cdf4ca..b048c17389 100644 --- a/test/DistEvents/DistDemoApp.MongoDbKafka/Program.cs +++ b/test/DistEvents/DistDemoApp.MongoDbKafka/Program.cs @@ -1,12 +1,57 @@ using System; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Serilog; +using Serilog.Events; namespace DistDemoApp { - class Program + public class Program { - static void Main(string[] args) + public static async Task Main(string[] args) { - Console.WriteLine("Hello World!"); + Log.Logger = new LoggerConfiguration() +#if DEBUG + .MinimumLevel.Debug() +#else + .MinimumLevel.Information() +#endif + .MinimumLevel.Override("Microsoft", LogEventLevel.Warning) + .Enrich.FromLogContext() + .WriteTo.Async(c => c.File("Logs/logs.txt")) + .WriteTo.Async(c => c.Console()) + .CreateLogger(); + + try + { + Log.Information("Starting console host."); + await CreateHostBuilder(args).RunConsoleAsync(); + return 0; + } + catch (Exception ex) + { + Log.Fatal(ex, "Host terminated unexpectedly!"); + return 1; + } + finally + { + Log.CloseAndFlush(); + } + } + + internal static IHostBuilder CreateHostBuilder(string[] args) => + Host.CreateDefaultBuilder(args) + .UseAutofac() + .UseSerilog() + .ConfigureAppConfiguration((context, config) => + { + //setup your additional configuration sources + }) + .ConfigureServices((hostContext, services) => + { + services.AddApplication(); + }); } -} \ No newline at end of file +} diff --git a/test/DistEvents/DistDemoApp.MongoDbKafka/TodoMongoDbContext.cs b/test/DistEvents/DistDemoApp.MongoDbKafka/TodoMongoDbContext.cs new file mode 100644 index 0000000000..a7f1b78f86 --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbKafka/TodoMongoDbContext.cs @@ -0,0 +1,26 @@ +using MongoDB.Driver; +using Volo.Abp.Data; +using Volo.Abp.MongoDB; +using Volo.Abp.MongoDB.DistributedEvents; + +namespace DistDemoApp +{ + [ConnectionStringName("Default")] + public class TodoMongoDbContext : AbpMongoDbContext, IHasEventOutbox, IHasEventInbox + { + public IMongoCollection TodoItems => Collection(); + public IMongoCollection TodoSummaries => Collection(); + + public IMongoCollection OutgoingEvents + { + get => Collection(); + set {} + } + public IMongoCollection IncomingEvents + { + get => Collection(); + set {} + } + } + +} diff --git a/test/DistEvents/DistDemoApp.MongoDbKafka/appsettings.json b/test/DistEvents/DistDemoApp.MongoDbKafka/appsettings.json new file mode 100644 index 0000000000..d8c528fe87 --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbKafka/appsettings.json @@ -0,0 +1,19 @@ +{ + "ConnectionStrings": { + "Default": "mongodb://localhost:27018,localhost:27019,localhost:27020/DistEventsDemo" + }, + "Kafka": { + "Connections": { + "Default": { + "BootstrapServers": "localhost:9092" + } + }, + "EventBus": { + "GroupId": "DistDemoApp", + "TopicName": "DistDemoTopic" + } + }, + "Redis": { + "Configuration": "127.0.0.1" + } +} diff --git a/test/DistEvents/DistDemoApp.MongoDbRebus/DistDemoApp.MongoDbRebus.csproj b/test/DistEvents/DistDemoApp.MongoDbRebus/DistDemoApp.MongoDbRebus.csproj new file mode 100644 index 0000000000..c102adb382 --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbRebus/DistDemoApp.MongoDbRebus.csproj @@ -0,0 +1,21 @@ + + + + Exe + net5.0 + DistDemoApp + + + + + + + + + + + Always + + + + diff --git a/test/DistEvents/DistDemoApp.MongoDbRebus/DistDemoAppMongoDbRebusModule.cs b/test/DistEvents/DistDemoApp.MongoDbRebus/DistDemoAppMongoDbRebusModule.cs new file mode 100644 index 0000000000..21dab7b9b5 --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbRebus/DistDemoAppMongoDbRebusModule.cs @@ -0,0 +1,53 @@ +using Microsoft.Extensions.DependencyInjection; +using Rebus.Persistence.InMem; +using Rebus.Transport.InMem; +using Volo.Abp.EventBus.Distributed; +using Volo.Abp.EventBus.Rebus; +using Volo.Abp.Modularity; +using Volo.Abp.MongoDB; +using Volo.Abp.MongoDB.DistributedEvents; + +namespace DistDemoApp +{ + [DependsOn( + typeof(AbpMongoDbModule), + typeof(AbpEventBusRebusModule), + typeof(DistDemoAppSharedModule) + )] + public class DistDemoAppMongoDbRebusModule : AbpModule + { + public override void PreConfigureServices(ServiceConfigurationContext context) + { + PreConfigure(options => + { + options.InputQueueName = "eventbus"; + options.Configurer = rebusConfigurer => + { + rebusConfigurer.Transport(t => t.UseInMemoryTransport(new InMemNetwork(), "eventbus")); + rebusConfigurer.Subscriptions(s => s.StoreInMemory()); + }; + }); + } + + public override void ConfigureServices(ServiceConfigurationContext context) + { + context.Services.AddMongoDbContext(options => + { + options.AddDefaultRepositories(); + }); + + Configure(options => + { + options.Outboxes.Configure(config => + { + config.UseMongoDbContext(); + }); + + options.Inboxes.Configure(config => + { + config.UseMongoDbContext(); + }); + }); + } + } +} diff --git a/test/DistEvents/DistDemoApp.MongoDbRebus/Program.cs b/test/DistEvents/DistDemoApp.MongoDbRebus/Program.cs new file mode 100644 index 0000000000..a79ad1b1bf --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbRebus/Program.cs @@ -0,0 +1,57 @@ +using System; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Serilog; +using Serilog.Events; + +namespace DistDemoApp +{ + public class Program + { + public static async Task Main(string[] args) + { + Log.Logger = new LoggerConfiguration() +#if DEBUG + .MinimumLevel.Debug() +#else + .MinimumLevel.Information() +#endif + .MinimumLevel.Override("Microsoft", LogEventLevel.Warning) + .Enrich.FromLogContext() + .WriteTo.Async(c => c.File("Logs/logs.txt")) + .WriteTo.Async(c => c.Console()) + .CreateLogger(); + + try + { + Log.Information("Starting console host."); + await CreateHostBuilder(args).RunConsoleAsync(); + return 0; + } + catch (Exception ex) + { + Log.Fatal(ex, "Host terminated unexpectedly!"); + return 1; + } + finally + { + Log.CloseAndFlush(); + } + + } + + internal static IHostBuilder CreateHostBuilder(string[] args) => + Host.CreateDefaultBuilder(args) + .UseAutofac() + .UseSerilog() + .ConfigureAppConfiguration((context, config) => + { + //setup your additional configuration sources + }) + .ConfigureServices((hostContext, services) => + { + services.AddApplication(); + }); + } +} diff --git a/test/DistEvents/DistDemoApp.MongoDbRebus/TodoMongoDbContext.cs b/test/DistEvents/DistDemoApp.MongoDbRebus/TodoMongoDbContext.cs new file mode 100644 index 0000000000..a7f1b78f86 --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbRebus/TodoMongoDbContext.cs @@ -0,0 +1,26 @@ +using MongoDB.Driver; +using Volo.Abp.Data; +using Volo.Abp.MongoDB; +using Volo.Abp.MongoDB.DistributedEvents; + +namespace DistDemoApp +{ + [ConnectionStringName("Default")] + public class TodoMongoDbContext : AbpMongoDbContext, IHasEventOutbox, IHasEventInbox + { + public IMongoCollection TodoItems => Collection(); + public IMongoCollection TodoSummaries => Collection(); + + public IMongoCollection OutgoingEvents + { + get => Collection(); + set {} + } + public IMongoCollection IncomingEvents + { + get => Collection(); + set {} + } + } + +} diff --git a/test/DistEvents/DistDemoApp.MongoDbRebus/appsettings.json b/test/DistEvents/DistDemoApp.MongoDbRebus/appsettings.json new file mode 100644 index 0000000000..d8c528fe87 --- /dev/null +++ b/test/DistEvents/DistDemoApp.MongoDbRebus/appsettings.json @@ -0,0 +1,19 @@ +{ + "ConnectionStrings": { + "Default": "mongodb://localhost:27018,localhost:27019,localhost:27020/DistEventsDemo" + }, + "Kafka": { + "Connections": { + "Default": { + "BootstrapServers": "localhost:9092" + } + }, + "EventBus": { + "GroupId": "DistDemoApp", + "TopicName": "DistDemoTopic" + } + }, + "Redis": { + "Configuration": "127.0.0.1" + } +} diff --git a/test/DistEvents/DistEventsDemo.sln b/test/DistEvents/DistEventsDemo.sln index 4e53ba6367..e6c3348a5f 100644 --- a/test/DistEvents/DistEventsDemo.sln +++ b/test/DistEvents/DistEventsDemo.sln @@ -6,6 +6,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DistDemoApp.MongoDbKafka", EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DistDemoApp.Shared", "DistDemoApp.Shared\DistDemoApp.Shared.csproj", "{C515F4E2-0ED3-4561-BC58-FC633B50E2EB}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "DistDemoApp.MongoDbRebus", "DistDemoApp.MongoDbRebus\DistDemoApp.MongoDbRebus.csproj", "{4FB63540-4CC5-4A7B-900B-F5FCD907456E}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -24,5 +26,9 @@ Global {C515F4E2-0ED3-4561-BC58-FC633B50E2EB}.Debug|Any CPU.Build.0 = Debug|Any CPU {C515F4E2-0ED3-4561-BC58-FC633B50E2EB}.Release|Any CPU.ActiveCfg = Release|Any CPU {C515F4E2-0ED3-4561-BC58-FC633B50E2EB}.Release|Any CPU.Build.0 = Release|Any CPU + {4FB63540-4CC5-4A7B-900B-F5FCD907456E}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {4FB63540-4CC5-4A7B-900B-F5FCD907456E}.Debug|Any CPU.Build.0 = Debug|Any CPU + {4FB63540-4CC5-4A7B-900B-F5FCD907456E}.Release|Any CPU.ActiveCfg = Release|Any CPU + {4FB63540-4CC5-4A7B-900B-F5FCD907456E}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection EndGlobal