mirror of https://github.com/abpframework/abp.git
Halil İbrahim Kalkan
4 years ago
22 changed files with 395 additions and 46 deletions
@ -1,29 +1,59 @@ |
|||
using System.Threading.Tasks; |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Guids; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class DbContextEventOutbox<TDbContext> : IEventOutbox |
|||
public class DbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
protected IDbContextProvider<TDbContext> DbContextProvider { get; } |
|||
protected IGuidGenerator GuidGenerator { get; } |
|||
|
|||
public DbContextEventOutbox( |
|||
IDbContextProvider<TDbContext> dbContextProvider, |
|||
IGuidGenerator guidGenerator) |
|||
IDbContextProvider<TDbContext> dbContextProvider) |
|||
{ |
|||
DbContextProvider = dbContextProvider; |
|||
GuidGenerator = guidGenerator; |
|||
} |
|||
|
|||
public async Task EnqueueAsync(string eventName, byte[] eventData) |
|||
[UnitOfWork] |
|||
public virtual async Task EnqueueAsync(OutgoingEventInfo outgoingEvent) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
dbContext.OutgoingEventRecords.Add( |
|||
new OutgoingEventRecord(GuidGenerator.Create(), eventName, eventData) |
|||
new OutgoingEventRecord(outgoingEvent) |
|||
); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
|
|||
var outgoingEventRecords = await dbContext |
|||
.OutgoingEventRecords |
|||
.AsNoTracking() |
|||
.OrderBy(x => x.CreationTime) |
|||
.Take(maxCount) |
|||
.ToListAsync(); |
|||
|
|||
return outgoingEventRecords |
|||
.Select(x => x.ToOutgoingEventInfo()) |
|||
.ToList(); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task DeleteAsync(Guid id) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
var outgoingEvent = await dbContext.OutgoingEventRecords.FindAsync(id); |
|||
if (outgoingEvent != null) |
|||
{ |
|||
dbContext.Remove(outgoingEvent); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IDbContextEventOutbox<TDbContext> : IEventOutbox |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,18 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public static class AbpDistributedEventBusExtensions |
|||
{ |
|||
public static IRawEventPublisher AsRawEventPublisher(this IDistributedEventBus eventBus) |
|||
{ |
|||
var rawPublisher = eventBus as IRawEventPublisher; |
|||
if (rawPublisher == null) |
|||
{ |
|||
throw new AbpException($"Given type ({eventBus.GetType().AssemblyQualifiedName}) should implement {nameof(IRawEventPublisher)}!"); |
|||
} |
|||
|
|||
return rawPublisher; |
|||
} |
|||
} |
|||
} |
|||
@ -1,12 +1,17 @@ |
|||
using Volo.Abp.Modularity; |
|||
using Volo.Abp.BackgroundWorkers; |
|||
using Volo.Abp.Modularity; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
[DependsOn( |
|||
typeof(AbpEventBusModule) |
|||
typeof(AbpEventBusModule), |
|||
typeof(AbpBackgroundWorkersModule) |
|||
)] |
|||
public class AbpEventBusBoxesModule : AbpModule |
|||
{ |
|||
|
|||
public override void OnApplicationInitialization(ApplicationInitializationContext context) |
|||
{ |
|||
context.AddBackgroundWorker<OutboxSenderManager>(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,11 @@ |
|||
using System.Threading.Tasks; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public interface IOutboxSender |
|||
{ |
|||
Task StartAsync(OutboxConfig outboxConfig); |
|||
Task StopAsync(); |
|||
} |
|||
} |
|||
@ -1,7 +1,78 @@ |
|||
namespace Volo.Abp.EventBus.Boxes |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Microsoft.Extensions.Logging; |
|||
using Microsoft.Extensions.Logging.Abstractions; |
|||
using Volo.Abp.DependencyInjection; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Threading; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public class OutboxSender |
|||
//TODO: use distributed lock!
|
|||
public class OutboxSender : IOutboxSender, ITransientDependency |
|||
{ |
|||
//Background worker & distributed lock
|
|||
protected IServiceProvider ServiceProvider { get; } |
|||
protected AbpTimer Timer { get; } |
|||
protected IDistributedEventBus DistributedEventBus { get; } |
|||
protected IEventOutbox Outbox { get; private set; } |
|||
public ILogger<OutboxSender> Logger { get; set; } |
|||
|
|||
public OutboxSender( |
|||
IServiceProvider serviceProvider, |
|||
AbpTimer timer, |
|||
IDistributedEventBus distributedEventBus) |
|||
{ |
|||
ServiceProvider = serviceProvider; |
|||
Timer = timer; |
|||
DistributedEventBus = distributedEventBus; |
|||
Timer.Period = 2000; //TODO: Config?
|
|||
Timer.Elapsed += TimerOnElapsed; |
|||
Logger = NullLogger<OutboxSender>.Instance; |
|||
} |
|||
|
|||
public virtual Task StartAsync(OutboxConfig outboxConfig) |
|||
{ |
|||
Outbox = (IEventOutbox)ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); |
|||
Timer.Start(); |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public virtual Task StopAsync() |
|||
{ |
|||
Timer.Stop(); |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
private void TimerOnElapsed(object sender, EventArgs e) |
|||
{ |
|||
AsyncHelper.RunSync(RunAsync); |
|||
} |
|||
|
|||
protected virtual async Task RunAsync() |
|||
{ |
|||
while (true) |
|||
{ |
|||
var waitingEvents = await Outbox.GetWaitingEventsAsync(100); |
|||
if (waitingEvents.Count <= 0) |
|||
{ |
|||
break; |
|||
} |
|||
|
|||
Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox."); |
|||
|
|||
foreach (var waitingEvent in waitingEvents) |
|||
{ |
|||
await DistributedEventBus |
|||
.AsRawEventPublisher() |
|||
.PublishRawAsync(waitingEvent.Id, waitingEvent.EventName, waitingEvent.EventData); |
|||
|
|||
await Outbox.DeleteAsync(waitingEvent.Id); |
|||
|
|||
Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}"); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.BackgroundWorkers; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public class OutboxSenderManager : IBackgroundWorker |
|||
{ |
|||
protected AbpDistributedEventBusOptions Options { get; } |
|||
protected IServiceProvider ServiceProvider { get; } |
|||
protected List<IOutboxSender> Senders { get; } |
|||
|
|||
public OutboxSenderManager( |
|||
IOptions<AbpDistributedEventBusOptions> options, |
|||
IServiceProvider serviceProvider) |
|||
{ |
|||
ServiceProvider = serviceProvider; |
|||
Options = options.Value; |
|||
Senders = new List<IOutboxSender>(); |
|||
} |
|||
|
|||
public async Task StartAsync(CancellationToken cancellationToken = default) |
|||
{ |
|||
foreach (var outboxConfig in Options.Outboxes.Values) |
|||
{ |
|||
if (outboxConfig.IsSendingEnabled) |
|||
{ |
|||
var sender = ServiceProvider.GetRequiredService<IOutboxSender>(); |
|||
await sender.StartAsync(outboxConfig); |
|||
Senders.Add(sender); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public async Task StopAsync(CancellationToken cancellationToken = default) |
|||
{ |
|||
foreach (var sender in Senders) |
|||
{ |
|||
await sender.StopAsync(); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,9 +1,15 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public interface IEventOutbox |
|||
{ |
|||
Task EnqueueAsync(string eventName, byte[] eventData); |
|||
Task EnqueueAsync(OutgoingEventInfo outgoingEvent); |
|||
|
|||
Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount); |
|||
|
|||
Task DeleteAsync(Guid id); |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public interface IRawEventPublisher |
|||
{ |
|||
Task PublishRawAsync( |
|||
Guid eventId, |
|||
string eventName, |
|||
byte[] eventData); |
|||
} |
|||
} |
|||
@ -0,0 +1,39 @@ |
|||
using System; |
|||
using Volo.Abp.Data; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class OutgoingEventInfo : IHasExtraProperties |
|||
{ |
|||
public static int MaxEventNameLength { get; set; } = 256; |
|||
|
|||
public ExtraPropertyDictionary ExtraProperties { get; protected set; } |
|||
|
|||
public Guid Id { get; } |
|||
|
|||
public string EventName { get; } |
|||
|
|||
public byte[] EventData { get; } |
|||
|
|||
public DateTime CreationTime { get; } |
|||
|
|||
protected OutgoingEventInfo() |
|||
{ |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public OutgoingEventInfo( |
|||
Guid id, |
|||
string eventName, |
|||
byte[] eventData, |
|||
DateTime creationTime) |
|||
{ |
|||
Id = id; |
|||
EventName = eventName; |
|||
EventData = eventData; |
|||
CreationTime = creationTime; |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
}} |
|||
Loading…
Reference in new issue