mirror of https://github.com/abpframework/abp.git
Halil İbrahim Kalkan
5 years ago
committed by
GitHub
GitHub
139 changed files with 3384 additions and 264 deletions
@ -0,0 +1,3 @@ |
|||
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd"> |
|||
<ConfigureAwait ContinueOnCapturedContext="false" /> |
|||
</Weavers> |
|||
@ -0,0 +1,30 @@ |
|||
<?xml version="1.0" encoding="utf-8"?> |
|||
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"> |
|||
<!-- This file was generated by Fody. Manual changes to this file will be lost when your project is rebuilt. --> |
|||
<xs:element name="Weavers"> |
|||
<xs:complexType> |
|||
<xs:all> |
|||
<xs:element name="ConfigureAwait" minOccurs="0" maxOccurs="1"> |
|||
<xs:complexType> |
|||
<xs:attribute name="ContinueOnCapturedContext" type="xs:boolean" /> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:all> |
|||
<xs:attribute name="VerifyAssembly" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'true' to run assembly verification (PEVerify) on the target assembly after all weavers have been executed.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="VerifyIgnoreCodes" type="xs:string"> |
|||
<xs:annotation> |
|||
<xs:documentation>A comma-separated list of error codes that can be safely ignored in assembly verification.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="GenerateXsd" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'false' to turn off automatic generation of the XML Schema file.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:schema> |
|||
@ -0,0 +1,25 @@ |
|||
<Project Sdk="Microsoft.NET.Sdk"> |
|||
|
|||
<Import Project="..\..\..\configureawait.props" /> |
|||
<Import Project="..\..\..\common.props" /> |
|||
|
|||
<PropertyGroup> |
|||
<TargetFramework>netstandard2.0</TargetFramework> |
|||
<AssemblyName>Volo.Abp.DistributedLocking</AssemblyName> |
|||
<PackageId>Volo.Abp.DistributedLocking</PackageId> |
|||
<AssetTargetFallback>$(AssetTargetFallback);portable-net45+win8+wp8+wpa81;</AssetTargetFallback> |
|||
<GenerateAssemblyConfigurationAttribute>false</GenerateAssemblyConfigurationAttribute> |
|||
<GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute> |
|||
<GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute> |
|||
<RootNamespace /> |
|||
</PropertyGroup> |
|||
|
|||
<ItemGroup> |
|||
<ProjectReference Include="..\Volo.Abp.Core\Volo.Abp.Core.csproj" /> |
|||
</ItemGroup> |
|||
|
|||
<ItemGroup> |
|||
<PackageReference Include="DistributedLock.Core" Version="1.0.2" /> |
|||
</ItemGroup> |
|||
|
|||
</Project> |
|||
@ -0,0 +1,9 @@ |
|||
using Volo.Abp.Modularity; |
|||
|
|||
namespace Volo.Abp.DistributedLocking |
|||
{ |
|||
public class AbpDistributedLockingModule : AbpModule |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class MySQLInboxConfigExtensions |
|||
{ |
|||
public static void UseMySQL<TDbContext>(this InboxConfig outboxConfig) |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(ISqlRawDbContextEventInbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class MySQLOutboxConfigExtensions |
|||
{ |
|||
public static void UseMySQL<TDbContext>(this OutboxConfig outboxConfig) |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(ISqlRawDbContextEventOutbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,8 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IOracleDbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,7 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IOracleDbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
} |
|||
} |
|||
@ -0,0 +1,49 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.EventBus.Boxes; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class OracleDbContextEventInbox<TDbContext> : DbContextEventInbox<TDbContext> , IOracleDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
public OracleDbContextEventInbox( |
|||
IDbContextProvider<TDbContext> dbContextProvider, |
|||
IClock clock, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
: base(dbContextProvider, clock, eventBusBoxesOptions) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task MarkAsProcessedAsync(Guid id) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"UPDATE \"{tableName}\" SET \"Processed\" = '1', \"ProcessedTime\" = TO_DATE('{Clock.Now}', 'yyyy-mm-dd hh24:mi:ss') WHERE \"Id\" = HEXTORAW('{GuidToOracleType(id)}')"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteOldEventsAsync() |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
var timeToKeepEvents = Clock.Now - EventBusBoxesOptions.WaitTimeToDeleteProcessedInboxEvents; |
|||
|
|||
var sql = $"DELETE FROM \"{tableName}\" WHERE \"Processed\" = '1' AND \"CreationTime\" < TO_DATE('{timeToKeepEvents}', 'yyyy-mm-dd hh24:mi:ss')"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
protected virtual string GuidToOracleType(Guid id) |
|||
{ |
|||
return BitConverter.ToString(id.ToByteArray()).Replace("-", "").ToUpper(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class OracleDbContextEventOutbox<TDbContext> : DbContextEventOutbox<TDbContext> , IOracleDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
public OracleDbContextEventOutbox(IDbContextProvider<TDbContext> dbContextProvider) |
|||
: base(dbContextProvider) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteAsync(Guid id) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.OutgoingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"DELETE FROM \"{tableName}\" WHERE \"Id\" = HEXTORAW('{GuidToOracleType(id)}')"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
protected virtual string GuidToOracleType(Guid id) |
|||
{ |
|||
return BitConverter.ToString(id.ToByteArray()).Replace("-", "").ToUpper(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class OracleInboxConfigExtensions |
|||
{ |
|||
public static void UseOracle<TDbContext>(this InboxConfig outboxConfig) |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IOracleDbContextEventInbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class OracleOutboxConfigExtensions |
|||
{ |
|||
public static void UseOracle<TDbContext>(this OutboxConfig outboxConfig) |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IOracleDbContextEventOutbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,8 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IOracleDbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,7 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IOracleDbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
} |
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.EventBus.Boxes; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class OracleDbContextEventInbox<TDbContext> : DbContextEventInbox<TDbContext> , IOracleDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
public OracleDbContextEventInbox( |
|||
IDbContextProvider<TDbContext> dbContextProvider, |
|||
IClock clock, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
: base(dbContextProvider, clock, eventBusBoxesOptions) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task MarkAsProcessedAsync(Guid id) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"UPDATE \"{tableName}\" SET \"Processed\" = '1', \"ProcessedTime\" = TO_DATE('{Clock.Now}', 'yyyy-mm-dd hh24:mi:ss') WHERE \"Id\" = HEXTORAW('{GuidToOracleType(id)}')"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteOldEventsAsync() |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
var timeToKeepEvents = Clock.Now - EventBusBoxesOptions.WaitTimeToDeleteProcessedInboxEvents; |
|||
|
|||
var sql = $"DELETE FROM \"{tableName}\" WHERE \"Processed\" = '1' AND \"CreationTime\" < TO_DATE('{timeToKeepEvents}', 'yyyy-mm-dd hh24:mi:ss')"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
protected virtual string GuidToOracleType(Guid id) |
|||
{ |
|||
return BitConverter.ToString(id.ToByteArray()).Replace("-", "").ToUpper(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class OracleDbContextEventOutbox<TDbContext> : DbContextEventOutbox<TDbContext> , IOracleDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
public OracleDbContextEventOutbox(IDbContextProvider<TDbContext> dbContextProvider) |
|||
: base(dbContextProvider) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteAsync(Guid id) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.OutgoingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"DELETE FROM \"{tableName}\" WHERE \"Id\" = HEXTORAW('{GuidToOracleType(id)}')"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
protected virtual string GuidToOracleType(Guid id) |
|||
{ |
|||
return BitConverter.ToString(id.ToByteArray()).Replace("-", "").ToUpper(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class OracleInboxConfigExtensions |
|||
{ |
|||
public static void UseOracle<TDbContext>(this InboxConfig outboxConfig) |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IOracleDbContextEventInbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class OracleOutboxConfigExtensions |
|||
{ |
|||
public static void UseOracle<TDbContext>(this OutboxConfig outboxConfig) |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IOracleDbContextEventOutbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,8 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IPostgreSqlDbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,7 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IPostgreSqlDbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
} |
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.EventBus.Boxes; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class PostgreSqlDbContextEventInbox<TDbContext> : DbContextEventInbox<TDbContext>, IPostgreSqlDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
public PostgreSqlDbContextEventInbox( |
|||
IDbContextProvider<TDbContext> dbContextProvider, |
|||
IClock clock, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
: base(dbContextProvider, clock, eventBusBoxesOptions) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task MarkAsProcessedAsync(Guid id) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"UPDATE \"{tableName}\" SET \"Processed\" = '1', \"ProcessedTime\" = '{Clock.Now}' WHERE \"Id\" = '{id}'"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteOldEventsAsync() |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
var timeToKeepEvents = Clock.Now - EventBusBoxesOptions.WaitTimeToDeleteProcessedInboxEvents; |
|||
|
|||
var sql = $"DELETE FROM \"{tableName}\" WHERE \"Processed\" = '1' AND \"CreationTime\" < '{timeToKeepEvents}'"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class PostgreSqlDbContextEventOutbox<TDbContext> : DbContextEventOutbox<TDbContext> , IPostgreSqlDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
public PostgreSqlDbContextEventOutbox(IDbContextProvider<TDbContext> dbContextProvider) : base(dbContextProvider) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteAsync(Guid id) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.OutgoingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"DELETE FROM \"{tableName}\" WHERE \"Id\" = '{id}'"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class PostgreSqlInboxConfigExtensions |
|||
{ |
|||
public static void UseNpgsql<TDbContext>(this InboxConfig outboxConfig) |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IPostgreSqlDbContextEventInbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class PostgreSqlOutboxConfigExtensions |
|||
{ |
|||
public static void UseNpgsql<TDbContext>(this OutboxConfig outboxConfig) |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IPostgreSqlDbContextEventOutbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class SqlServerInboxConfigExtensions |
|||
{ |
|||
public static void UseSqlServer<TDbContext>(this InboxConfig outboxConfig) |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(ISqlRawDbContextEventInbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class SqlServerOutboxConfigExtensions |
|||
{ |
|||
public static void UseSqlServer<TDbContext>(this OutboxConfig outboxConfig) |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(ISqlRawDbContextEventOutbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class SqliteInboxConfigExtensions |
|||
{ |
|||
public static void UseSqlite<TDbContext>(this InboxConfig outboxConfig) |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(ISqlRawDbContextEventInbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class SqliteOutboxConfigExtensions |
|||
{ |
|||
public static void UseSqlite<TDbContext>(this OutboxConfig outboxConfig) |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(ISqlRawDbContextEventOutbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,89 @@ |
|||
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.Boxes; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class DbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
protected IDbContextProvider<TDbContext> DbContextProvider { get; } |
|||
protected AbpEventBusBoxesOptions EventBusBoxesOptions { get; } |
|||
protected IClock Clock { get; } |
|||
|
|||
public DbContextEventInbox( |
|||
IDbContextProvider<TDbContext> dbContextProvider, |
|||
IClock clock, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
{ |
|||
DbContextProvider = dbContextProvider; |
|||
Clock = clock; |
|||
EventBusBoxesOptions = eventBusBoxesOptions.Value; |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task EnqueueAsync(IncomingEventInfo incomingEvent) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
|
|||
dbContext.IncomingEvents.Add( |
|||
new IncomingEventRecord(incomingEvent) |
|||
); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task<List<IncomingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
|
|||
var outgoingEventRecords = await dbContext |
|||
.IncomingEvents |
|||
.AsNoTracking() |
|||
.Where(x => !x.Processed) |
|||
.OrderBy(x => x.CreationTime) |
|||
.Take(maxCount) |
|||
.ToListAsync(cancellationToken: cancellationToken); |
|||
|
|||
return outgoingEventRecords |
|||
.Select(x => x.ToIncomingEventInfo()) |
|||
.ToList(); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task MarkAsProcessedAsync(Guid id) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var incomingEvent = await dbContext.IncomingEvents.FindAsync(id); |
|||
if (incomingEvent != null) |
|||
{ |
|||
incomingEvent.MarkAsProcessed(Clock.Now); |
|||
} |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task<bool> ExistsByMessageIdAsync(string messageId) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
return await dbContext.IncomingEvents.AnyAsync(x => x.MessageId == messageId); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task DeleteOldEventsAsync() |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var timeToKeepEvents = Clock.Now - EventBusBoxesOptions.WaitTimeToDeleteProcessedInboxEvents; |
|||
var oldEvents = await dbContext.IncomingEvents |
|||
.Where(x => x.Processed && x.CreationTime < timeToKeepEvents) |
|||
.ToListAsync(); |
|||
dbContext.IncomingEvents.RemoveRange(oldEvents); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class DbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
protected IDbContextProvider<TDbContext> DbContextProvider { get; } |
|||
|
|||
public DbContextEventOutbox( |
|||
IDbContextProvider<TDbContext> dbContextProvider) |
|||
{ |
|||
DbContextProvider = dbContextProvider; |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task EnqueueAsync(OutgoingEventInfo outgoingEvent) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
dbContext.OutgoingEvents.Add( |
|||
new OutgoingEventRecord(outgoingEvent) |
|||
); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task<List<OutgoingEventInfo>> 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(cancellationToken: cancellationToken); |
|||
|
|||
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.OutgoingEvents.FindAsync(id); |
|||
if (outgoingEvent != null) |
|||
{ |
|||
dbContext.Remove(outgoingEvent); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class EfCoreInboxConfigExtensions |
|||
{ |
|||
public static void UseDbContext<TDbContext>(this InboxConfig outboxConfig) |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IDbContextEventInbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class EfCoreOutboxConfigExtensions |
|||
{ |
|||
public static void UseDbContext<TDbContext>(this OutboxConfig outboxConfig) |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
outboxConfig.ImplementationType = typeof(IDbContextEventOutbox<TDbContext>); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,24 @@ |
|||
using JetBrains.Annotations; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.Data; |
|||
using Volo.Abp.EntityFrameworkCore.Modeling; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class EventInboxDbContextModelBuilderExtensions |
|||
{ |
|||
public static void ConfigureEventInbox([NotNull] this ModelBuilder builder) |
|||
{ |
|||
builder.Entity<IncomingEventRecord>(b => |
|||
{ |
|||
b.ToTable(AbpCommonDbProperties.DbTablePrefix + "EventInbox", AbpCommonDbProperties.DbSchema); |
|||
b.ConfigureByConvention(); |
|||
b.Property(x => x.EventName).IsRequired().HasMaxLength(IncomingEventRecord.MaxEventNameLength); |
|||
b.Property(x => x.EventData).IsRequired(); |
|||
|
|||
b.HasIndex(x => new { x.Processed, x.CreationTime }); |
|||
b.HasIndex(x => x.MessageId); |
|||
}); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
using JetBrains.Annotations; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.Data; |
|||
using Volo.Abp.EntityFrameworkCore.Modeling; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public static class EventOutboxDbContextModelBuilderExtensions |
|||
{ |
|||
public static void ConfigureEventOutbox([NotNull] this ModelBuilder builder) |
|||
{ |
|||
builder.Entity<OutgoingEventRecord>(b => |
|||
{ |
|||
b.ToTable(AbpCommonDbProperties.DbTablePrefix + "EventOutbox", AbpCommonDbProperties.DbSchema); |
|||
b.ConfigureByConvention(); |
|||
b.Property(x => x.EventName).IsRequired().HasMaxLength(OutgoingEventRecord.MaxEventNameLength); |
|||
b.Property(x => x.EventData).IsRequired(); |
|||
|
|||
b.HasIndex(x => x.CreationTime); |
|||
}); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IDbContextEventInbox<TDbContext> : IEventInbox |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -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,9 @@ |
|||
using Microsoft.EntityFrameworkCore; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IHasEventInbox : IEfCoreDbContext |
|||
{ |
|||
DbSet<IncomingEventRecord> IncomingEvents { get; set; } |
|||
} |
|||
} |
|||
@ -0,0 +1,9 @@ |
|||
using Microsoft.EntityFrameworkCore; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface IHasEventOutbox : IEfCoreDbContext |
|||
{ |
|||
DbSet<OutgoingEventRecord> OutgoingEvents { get; set; } |
|||
} |
|||
} |
|||
@ -0,0 +1,7 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface ISqlRawDbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
} |
|||
} |
|||
@ -0,0 +1,7 @@ |
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public interface ISqlRawDbContextEventOutbox<TDbContext> : IDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
} |
|||
} |
|||
@ -0,0 +1,66 @@ |
|||
using System; |
|||
using Volo.Abp.Auditing; |
|||
using Volo.Abp.Data; |
|||
using Volo.Abp.Domain.Entities; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class IncomingEventRecord : |
|||
BasicAggregateRoot<Guid>, |
|||
IHasExtraProperties, |
|||
IHasCreationTime |
|||
{ |
|||
public static int MaxEventNameLength { get; set; } = 256; |
|||
|
|||
public ExtraPropertyDictionary ExtraProperties { get; private set; } |
|||
|
|||
public string MessageId { get; private set; } |
|||
|
|||
public string EventName { get; private set; } |
|||
|
|||
public byte[] EventData { get; private set; } |
|||
|
|||
public DateTime CreationTime { get; private set; } |
|||
|
|||
public bool Processed { get; set; } |
|||
|
|||
public DateTime? ProcessedTime { get; set; } |
|||
|
|||
protected IncomingEventRecord() |
|||
{ |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public IncomingEventRecord( |
|||
IncomingEventInfo eventInfo) |
|||
: base(eventInfo.Id) |
|||
{ |
|||
MessageId = eventInfo.MessageId; |
|||
EventName = eventInfo.EventName; |
|||
EventData = eventInfo.EventData; |
|||
CreationTime = eventInfo.CreationTime; |
|||
|
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public IncomingEventInfo ToIncomingEventInfo() |
|||
{ |
|||
return new IncomingEventInfo( |
|||
Id, |
|||
MessageId, |
|||
EventName, |
|||
EventData, |
|||
CreationTime |
|||
); |
|||
} |
|||
|
|||
public void MarkAsProcessed(DateTime processedTime) |
|||
{ |
|||
Processed = true; |
|||
ProcessedTime = processedTime; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,52 @@ |
|||
using System; |
|||
using Volo.Abp.Auditing; |
|||
using Volo.Abp.Data; |
|||
using Volo.Abp.Domain.Entities; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class OutgoingEventRecord : |
|||
BasicAggregateRoot<Guid>, |
|||
IHasExtraProperties, |
|||
IHasCreationTime |
|||
{ |
|||
public static int MaxEventNameLength { get; set; } = 256; |
|||
|
|||
public ExtraPropertyDictionary ExtraProperties { get; private set; } |
|||
|
|||
public string EventName { get; private set; } |
|||
|
|||
public byte[] EventData { get; private set; } |
|||
|
|||
public DateTime CreationTime { get; private set; } |
|||
|
|||
protected OutgoingEventRecord() |
|||
{ |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public OutgoingEventRecord( |
|||
OutgoingEventInfo eventInfo) |
|||
: base(eventInfo.Id) |
|||
{ |
|||
EventName = eventInfo.EventName; |
|||
EventData = eventInfo.EventData; |
|||
CreationTime = eventInfo.CreationTime; |
|||
|
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public OutgoingEventInfo ToOutgoingEventInfo() |
|||
{ |
|||
return new OutgoingEventInfo( |
|||
Id, |
|||
EventName, |
|||
EventData, |
|||
CreationTime |
|||
); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.EventBus.Boxes; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class SqlRawDbContextEventInbox<TDbContext> : DbContextEventInbox<TDbContext> , ISqlRawDbContextEventInbox<TDbContext> |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
public SqlRawDbContextEventInbox( |
|||
IDbContextProvider<TDbContext> dbContextProvider, |
|||
IClock clock, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
: base(dbContextProvider, clock, eventBusBoxesOptions) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task MarkAsProcessedAsync(Guid id) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"UPDATE {tableName} SET Processed = '1', ProcessedTime = '{Clock.Now}' WHERE Id = '{id.ToString().ToUpper()}'"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteOldEventsAsync() |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
var timeToKeepEvents = Clock.Now - EventBusBoxesOptions.WaitTimeToDeleteProcessedInboxEvents; |
|||
|
|||
var sql = $"DELETE FROM {tableName} WHERE Processed = '1' AND CreationTime < '{timeToKeepEvents}'"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.EntityFrameworkCore; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EntityFrameworkCore.DistributedEvents |
|||
{ |
|||
public class SqlRawDbContextEventOutbox<TDbContext> : DbContextEventOutbox<TDbContext> , ISqlRawDbContextEventOutbox<TDbContext> |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
public SqlRawDbContextEventOutbox(IDbContextProvider<TDbContext> dbContextProvider) |
|||
: base(dbContextProvider) |
|||
{ |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public override async Task DeleteAsync(Guid id) |
|||
{ |
|||
var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync(); |
|||
var tableName = dbContext.OutgoingEvents.EntityType.GetSchemaQualifiedTableName(); |
|||
|
|||
var sql = $"DELETE FROM {tableName} WHERE Id = '{id.ToString().ToUpper()}'"; |
|||
await dbContext.Database.ExecuteSqlRawAsync(sql); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,3 @@ |
|||
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd"> |
|||
<ConfigureAwait ContinueOnCapturedContext="false" /> |
|||
</Weavers> |
|||
@ -0,0 +1,30 @@ |
|||
<?xml version="1.0" encoding="utf-8"?> |
|||
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"> |
|||
<!-- This file was generated by Fody. Manual changes to this file will be lost when your project is rebuilt. --> |
|||
<xs:element name="Weavers"> |
|||
<xs:complexType> |
|||
<xs:all> |
|||
<xs:element name="ConfigureAwait" minOccurs="0" maxOccurs="1"> |
|||
<xs:complexType> |
|||
<xs:attribute name="ContinueOnCapturedContext" type="xs:boolean" /> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:all> |
|||
<xs:attribute name="VerifyAssembly" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'true' to run assembly verification (PEVerify) on the target assembly after all weavers have been executed.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="VerifyIgnoreCodes" type="xs:string"> |
|||
<xs:annotation> |
|||
<xs:documentation>A comma-separated list of error codes that can be safely ignored in assembly verification.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
<xs:attribute name="GenerateXsd" type="xs:boolean"> |
|||
<xs:annotation> |
|||
<xs:documentation>'false' to turn off automatic generation of the XML Schema file.</xs:documentation> |
|||
</xs:annotation> |
|||
</xs:attribute> |
|||
</xs:complexType> |
|||
</xs:element> |
|||
</xs:schema> |
|||
@ -0,0 +1,23 @@ |
|||
<Project Sdk="Microsoft.NET.Sdk"> |
|||
|
|||
<Import Project="..\..\..\configureawait.props" /> |
|||
<Import Project="..\..\..\common.props" /> |
|||
|
|||
<PropertyGroup> |
|||
<TargetFramework>netstandard2.0</TargetFramework> |
|||
<AssemblyName>Volo.Abp.EventBus.Boxes</AssemblyName> |
|||
<PackageId>Volo.Abp.EventBus.Boxes</PackageId> |
|||
<AssetTargetFallback>$(AssetTargetFallback);portable-net45+win8+wp8+wpa81;</AssetTargetFallback> |
|||
<GenerateAssemblyConfigurationAttribute>false</GenerateAssemblyConfigurationAttribute> |
|||
<GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute> |
|||
<GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute> |
|||
<RootNamespace /> |
|||
</PropertyGroup> |
|||
|
|||
<ItemGroup> |
|||
<ProjectReference Include="..\Volo.Abp.BackgroundWorkers\Volo.Abp.BackgroundWorkers.csproj" /> |
|||
<ProjectReference Include="..\Volo.Abp.DistributedLocking\Volo.Abp.DistributedLocking.csproj" /> |
|||
<ProjectReference Include="..\Volo.Abp.EventBus\Volo.Abp.EventBus.csproj" /> |
|||
</ItemGroup> |
|||
|
|||
</Project> |
|||
@ -0,0 +1,18 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public static class AbpDistributedEventBusExtensions |
|||
{ |
|||
public static ISupportsEventBoxes AsSupportsEventBoxes(this IDistributedEventBus eventBus) |
|||
{ |
|||
var supportsEventBoxes = eventBus as ISupportsEventBoxes; |
|||
if (supportsEventBoxes == null) |
|||
{ |
|||
throw new AbpException($"Given type ({eventBus.GetType().AssemblyQualifiedName}) should implement {nameof(ISupportsEventBoxes)}!"); |
|||
} |
|||
|
|||
return supportsEventBoxes; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,18 @@ |
|||
using Volo.Abp.BackgroundWorkers; |
|||
using Volo.Abp.Modularity; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
[DependsOn( |
|||
typeof(AbpEventBusModule), |
|||
typeof(AbpBackgroundWorkersModule) |
|||
)] |
|||
public class AbpEventBusBoxesModule : AbpModule |
|||
{ |
|||
public override void OnApplicationInitialization(ApplicationInitializationContext context) |
|||
{ |
|||
context.AddBackgroundWorker<OutboxSenderManager>(); |
|||
context.AddBackgroundWorker<InboxProcessManager>(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
using System; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public class AbpEventBusBoxesOptions |
|||
{ |
|||
/// <summary>
|
|||
/// Default: 6 hours
|
|||
/// </summary>
|
|||
public TimeSpan CleanOldEventTimeIntervalSpan { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// Default: 1000
|
|||
/// </summary>
|
|||
public int InboxWaitingEventMaxCount { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// Default: 1000
|
|||
/// </summary>
|
|||
public int OutboxWaitingEventMaxCount { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// Period time of <see cref="InboxProcessor"/> and <see cref="OutboxSender"/>
|
|||
/// Default: 2 seconds
|
|||
/// </summary>
|
|||
public TimeSpan PeriodTimeSpan { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// Default: 15 seconds
|
|||
/// </summary>
|
|||
public TimeSpan DistributedLockWaitDuration { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// Default: 2 hours
|
|||
/// </summary>
|
|||
public TimeSpan WaitTimeToDeleteProcessedInboxEvents { get; set; } |
|||
|
|||
public AbpEventBusBoxesOptions() |
|||
{ |
|||
CleanOldEventTimeIntervalSpan = TimeSpan.FromHours(6); |
|||
InboxWaitingEventMaxCount = 1000; |
|||
OutboxWaitingEventMaxCount = 1000; |
|||
PeriodTimeSpan = TimeSpan.FromSeconds(2); |
|||
DistributedLockWaitDuration = TimeSpan.FromSeconds(15); |
|||
WaitTimeToDeleteProcessedInboxEvents = TimeSpan.FromHours(2); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public interface IInboxProcessor |
|||
{ |
|||
Task StartAsync(InboxConfig inboxConfig, CancellationToken cancellationToken = default); |
|||
|
|||
Task StopAsync(CancellationToken cancellationToken = default); |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public interface IOutboxSender |
|||
{ |
|||
Task StartAsync(OutboxConfig outboxConfig, CancellationToken cancellationToken = default); |
|||
|
|||
Task StopAsync(CancellationToken cancellationToken = default); |
|||
} |
|||
} |
|||
@ -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 InboxProcessManager : IBackgroundWorker |
|||
{ |
|||
protected AbpDistributedEventBusOptions Options { get; } |
|||
protected IServiceProvider ServiceProvider { get; } |
|||
protected List<IInboxProcessor> Processors { get; } |
|||
|
|||
public InboxProcessManager( |
|||
IOptions<AbpDistributedEventBusOptions> options, |
|||
IServiceProvider serviceProvider) |
|||
{ |
|||
ServiceProvider = serviceProvider; |
|||
Options = options.Value; |
|||
Processors = new List<IInboxProcessor>(); |
|||
} |
|||
|
|||
public async Task StartAsync(CancellationToken cancellationToken = default) |
|||
{ |
|||
foreach (var inboxConfig in Options.Inboxes.Values) |
|||
{ |
|||
if (inboxConfig.IsProcessingEnabled) |
|||
{ |
|||
var processor = ServiceProvider.GetRequiredService<IInboxProcessor>(); |
|||
await processor.StartAsync(inboxConfig, cancellationToken); |
|||
Processors.Add(processor); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public async Task StopAsync(CancellationToken cancellationToken = default) |
|||
{ |
|||
foreach (var processor in Processors) |
|||
{ |
|||
await processor.StopAsync(cancellationToken); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,140 @@ |
|||
using System; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
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; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public class InboxProcessor : IInboxProcessor, ITransientDependency |
|||
{ |
|||
protected IServiceProvider ServiceProvider { get; } |
|||
protected AbpAsyncTimer Timer { get; } |
|||
protected IDistributedEventBus DistributedEventBus { get; } |
|||
protected IDistributedLockProvider DistributedLockProvider { get; } |
|||
protected IUnitOfWorkManager UnitOfWorkManager { get; } |
|||
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; } |
|||
|
|||
protected string DistributedLockName => "Inbox_" + InboxConfig.Name; |
|||
public ILogger<InboxProcessor> Logger { get; set; } |
|||
protected CancellationTokenSource StoppingTokenSource { get; } |
|||
protected CancellationToken StoppingToken { get; } |
|||
|
|||
public InboxProcessor( |
|||
IServiceProvider serviceProvider, |
|||
AbpAsyncTimer timer, |
|||
IDistributedEventBus distributedEventBus, |
|||
IDistributedLockProvider distributedLockProvider, |
|||
IUnitOfWorkManager unitOfWorkManager, |
|||
IClock clock, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
{ |
|||
ServiceProvider = serviceProvider; |
|||
Timer = timer; |
|||
DistributedEventBus = distributedEventBus; |
|||
DistributedLockProvider = distributedLockProvider; |
|||
UnitOfWorkManager = unitOfWorkManager; |
|||
Clock = clock; |
|||
EventBusBoxesOptions = eventBusBoxesOptions.Value; |
|||
Timer.Period = EventBusBoxesOptions.PeriodTimeSpan.Milliseconds; |
|||
Timer.Elapsed += TimerOnElapsed; |
|||
Logger = NullLogger<InboxProcessor>.Instance; |
|||
StoppingTokenSource = new CancellationTokenSource(); |
|||
StoppingToken = StoppingTokenSource.Token; |
|||
} |
|||
|
|||
private async Task TimerOnElapsed(AbpAsyncTimer arg) |
|||
{ |
|||
await RunAsync(); |
|||
} |
|||
|
|||
public Task StartAsync(InboxConfig inboxConfig, CancellationToken cancellationToken = default) |
|||
{ |
|||
InboxConfig = inboxConfig; |
|||
Inbox = (IEventInbox)ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); |
|||
Timer.Start(cancellationToken); |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public Task StopAsync(CancellationToken cancellationToken = default) |
|||
{ |
|||
StoppingTokenSource.Cancel(); |
|||
Timer.Stop(cancellationToken); |
|||
StoppingTokenSource.Dispose(); |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
protected virtual async Task RunAsync() |
|||
{ |
|||
if (StoppingToken.IsCancellationRequested) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
await using (var handle = await DistributedLockProvider.TryAcquireLockAsync(DistributedLockName, cancellationToken: StoppingToken)) |
|||
{ |
|||
if (handle != null) |
|||
{ |
|||
await DeleteOldEventsAsync(); |
|||
|
|||
while (true) |
|||
{ |
|||
var waitingEvents = await Inbox.GetWaitingEventsAsync(EventBusBoxesOptions.InboxWaitingEventMaxCount, StoppingToken); |
|||
if (waitingEvents.Count <= 0) |
|||
{ |
|||
break; |
|||
} |
|||
|
|||
Logger.LogInformation($"Found {waitingEvents.Count} events in the inbox."); |
|||
|
|||
foreach (var waitingEvent in waitingEvents) |
|||
{ |
|||
using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true)) |
|||
{ |
|||
await DistributedEventBus |
|||
.AsSupportsEventBoxes() |
|||
.ProcessFromInboxAsync(waitingEvent, InboxConfig); |
|||
|
|||
await Inbox.MarkAsProcessedAsync(waitingEvent.Id); |
|||
|
|||
await uow.CompleteAsync(); |
|||
} |
|||
|
|||
Logger.LogInformation($"Processed the incoming event with id = {waitingEvent.Id:N}"); |
|||
} |
|||
} |
|||
} |
|||
else |
|||
{ |
|||
Logger.LogDebug("Could not obtain the distributed lock: " + DistributedLockName); |
|||
await TaskDelayHelper.DelayAsync(EventBusBoxesOptions.DistributedLockWaitDuration.Milliseconds, StoppingToken); |
|||
} |
|||
} |
|||
} |
|||
|
|||
protected virtual async Task DeleteOldEventsAsync() |
|||
{ |
|||
if (LastCleanTime != null && LastCleanTime + EventBusBoxesOptions.CleanOldEventTimeIntervalSpan > Clock.Now) |
|||
{ |
|||
return; |
|||
} |
|||
|
|||
await Inbox.DeleteOldEventsAsync(); |
|||
|
|||
LastCleanTime = Clock.Now; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,108 @@ |
|||
using System; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
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; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
public class OutboxSender : IOutboxSender, ITransientDependency |
|||
{ |
|||
protected IServiceProvider ServiceProvider { get; } |
|||
protected AbpAsyncTimer Timer { get; } |
|||
protected IDistributedEventBus DistributedEventBus { get; } |
|||
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<OutboxSender> Logger { get; set; } |
|||
|
|||
protected CancellationTokenSource StoppingTokenSource { get; } |
|||
protected CancellationToken StoppingToken { get; } |
|||
|
|||
public OutboxSender( |
|||
IServiceProvider serviceProvider, |
|||
AbpAsyncTimer timer, |
|||
IDistributedEventBus distributedEventBus, |
|||
IDistributedLockProvider distributedLockProvider, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
{ |
|||
ServiceProvider = serviceProvider; |
|||
Timer = timer; |
|||
DistributedEventBus = distributedEventBus; |
|||
DistributedLockProvider = distributedLockProvider; |
|||
EventBusBoxesOptions = eventBusBoxesOptions.Value; |
|||
Timer.Period = EventBusBoxesOptions.PeriodTimeSpan.Milliseconds; |
|||
Timer.Elapsed += TimerOnElapsed; |
|||
Logger = NullLogger<OutboxSender>.Instance; |
|||
StoppingTokenSource = new CancellationTokenSource(); |
|||
StoppingToken = StoppingTokenSource.Token; |
|||
} |
|||
|
|||
public virtual Task StartAsync(OutboxConfig outboxConfig, CancellationToken cancellationToken = default) |
|||
{ |
|||
OutboxConfig = outboxConfig; |
|||
Outbox = (IEventOutbox)ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); |
|||
Timer.Start(cancellationToken); |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public virtual Task StopAsync(CancellationToken cancellationToken = default) |
|||
{ |
|||
StoppingTokenSource.Cancel(); |
|||
Timer.Stop(cancellationToken); |
|||
StoppingTokenSource.Dispose(); |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
private async Task TimerOnElapsed(AbpAsyncTimer arg) |
|||
{ |
|||
await RunAsync(); |
|||
} |
|||
|
|||
protected virtual async Task RunAsync() |
|||
{ |
|||
await using (var handle = await DistributedLockProvider.TryAcquireLockAsync(DistributedLockName, cancellationToken: StoppingToken)) |
|||
{ |
|||
if (handle != null) |
|||
{ |
|||
while (true) |
|||
{ |
|||
var waitingEvents = await Outbox.GetWaitingEventsAsync(EventBusBoxesOptions.OutboxWaitingEventMaxCount, StoppingToken); |
|||
if (waitingEvents.Count <= 0) |
|||
{ |
|||
break; |
|||
} |
|||
|
|||
Logger.LogInformation($"Found {waitingEvents.Count} events in the outbox."); |
|||
|
|||
foreach (var waitingEvent in waitingEvents) |
|||
{ |
|||
await DistributedEventBus |
|||
.AsSupportsEventBoxes() |
|||
.PublishFromOutboxAsync( |
|||
waitingEvent, |
|||
OutboxConfig |
|||
); |
|||
|
|||
await Outbox.DeleteAsync(waitingEvent.Id); |
|||
Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}"); |
|||
} |
|||
} |
|||
} |
|||
else |
|||
{ |
|||
Logger.LogDebug("Could not obtain the distributed lock: " + DistributedLockName); |
|||
await TaskDelayHelper.DelayAsync(EventBusBoxesOptions.DistributedLockWaitDuration.Milliseconds, StoppingToken); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -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, cancellationToken); |
|||
Senders.Add(sender); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public async Task StopAsync(CancellationToken cancellationToken = default) |
|||
{ |
|||
foreach (var sender in Senders) |
|||
{ |
|||
await sender.StopAsync(cancellationToken); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus.Boxes |
|||
{ |
|||
internal static class TaskDelayHelper |
|||
{ |
|||
public static async Task DelayAsync(int milliseconds, CancellationToken cancellationToken = default) |
|||
{ |
|||
try |
|||
{ |
|||
await Task.Delay(milliseconds, cancellationToken); |
|||
} |
|||
catch (TaskCanceledException) |
|||
{ |
|||
return; |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,13 @@ |
|||
using System; |
|||
|
|||
namespace Volo.Abp.EventBus.Rebus |
|||
{ |
|||
public interface IRebusSerializer |
|||
{ |
|||
byte[] Serialize(object obj); |
|||
|
|||
object Deserialize(byte[] value, Type type); |
|||
|
|||
T Deserialize<T>(byte[] value); |
|||
} |
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
using System; |
|||
using System.Text; |
|||
using Volo.Abp.DependencyInjection; |
|||
using Volo.Abp.Json; |
|||
|
|||
namespace Volo.Abp.EventBus.Rebus |
|||
{ |
|||
public class Utf8JsonRebusSerializer : IRebusSerializer, ITransientDependency |
|||
{ |
|||
private readonly IJsonSerializer _jsonSerializer; |
|||
|
|||
public Utf8JsonRebusSerializer(IJsonSerializer jsonSerializer) |
|||
{ |
|||
_jsonSerializer = jsonSerializer; |
|||
} |
|||
|
|||
public byte[] Serialize(object obj) |
|||
{ |
|||
return Encoding.UTF8.GetBytes(_jsonSerializer.Serialize(obj)); |
|||
} |
|||
|
|||
public object Deserialize(byte[] value, Type type) |
|||
{ |
|||
return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value)); |
|||
} |
|||
|
|||
public T Deserialize<T>(byte[] value) |
|||
{ |
|||
return _jsonSerializer.Deserialize<T>(Encoding.UTF8.GetString(value)); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,166 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Threading.Tasks; |
|||
using Microsoft.Extensions.DependencyInjection; |
|||
using Microsoft.Extensions.Options; |
|||
using Volo.Abp.Guids; |
|||
using Volo.Abp.MultiTenancy; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventBus, ISupportsEventBoxes |
|||
{ |
|||
protected IGuidGenerator GuidGenerator { get; } |
|||
protected IClock Clock { get; } |
|||
protected AbpDistributedEventBusOptions AbpDistributedEventBusOptions { get; } |
|||
|
|||
protected DistributedEventBusBase( |
|||
IServiceScopeFactory serviceScopeFactory, |
|||
ICurrentTenant currentTenant, |
|||
IUnitOfWorkManager unitOfWorkManager, |
|||
IEventErrorHandler errorHandler, |
|||
IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions, |
|||
IGuidGenerator guidGenerator, |
|||
IClock clock |
|||
) : base( |
|||
serviceScopeFactory, |
|||
currentTenant, |
|||
unitOfWorkManager, |
|||
errorHandler) |
|||
{ |
|||
GuidGenerator = guidGenerator; |
|||
Clock = clock; |
|||
AbpDistributedEventBusOptions = abpDistributedEventBusOptions.Value; |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
return Subscribe(typeof(TEvent), handler); |
|||
} |
|||
|
|||
public override Task PublishAsync(Type eventType, object eventData, bool onUnitOfWorkComplete = true) |
|||
{ |
|||
return PublishAsync(eventType, eventData, onUnitOfWorkComplete, useOutbox: true); |
|||
} |
|||
|
|||
public Task PublishAsync<TEvent>( |
|||
TEvent eventData, |
|||
bool onUnitOfWorkComplete = true, |
|||
bool useOutbox = true) |
|||
where TEvent : class |
|||
{ |
|||
return PublishAsync(typeof(TEvent), eventData, onUnitOfWorkComplete, useOutbox); |
|||
} |
|||
|
|||
public async Task PublishAsync( |
|||
Type eventType, |
|||
object eventData, |
|||
bool onUnitOfWorkComplete = true, |
|||
bool useOutbox = true) |
|||
{ |
|||
if (onUnitOfWorkComplete && UnitOfWorkManager.Current != null) |
|||
{ |
|||
AddToUnitOfWork( |
|||
UnitOfWorkManager.Current, |
|||
new UnitOfWorkEventRecord(eventType, eventData, EventOrderGenerator.GetNext(), useOutbox) |
|||
); |
|||
return; |
|||
} |
|||
|
|||
if (useOutbox) |
|||
{ |
|||
if (await AddToOutboxAsync(eventType, eventData)) |
|||
{ |
|||
return; |
|||
} |
|||
} |
|||
|
|||
await PublishToEventBusAsync(eventType, eventData); |
|||
} |
|||
|
|||
public abstract Task PublishFromOutboxAsync( |
|||
OutgoingEventInfo outgoingEvent, |
|||
OutboxConfig outboxConfig |
|||
); |
|||
|
|||
public abstract Task ProcessFromInboxAsync( |
|||
IncomingEventInfo incomingEvent, |
|||
InboxConfig inboxConfig); |
|||
|
|||
private async Task<bool> AddToOutboxAsync(Type eventType, object eventData) |
|||
{ |
|||
var unitOfWork = UnitOfWorkManager.Current; |
|||
if (unitOfWork == null) |
|||
{ |
|||
return false; |
|||
} |
|||
|
|||
foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values) |
|||
{ |
|||
if (outboxConfig.Selector == null || outboxConfig.Selector(eventType)) |
|||
{ |
|||
var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); |
|||
var eventName = EventNameAttribute.GetNameOrDefault(eventType); |
|||
await eventOutbox.EnqueueAsync( |
|||
new OutgoingEventInfo( |
|||
GuidGenerator.Create(), |
|||
eventName, |
|||
Serialize(eventData), |
|||
Clock.Now |
|||
) |
|||
); |
|||
return true; |
|||
} |
|||
} |
|||
|
|||
return false; |
|||
} |
|||
|
|||
protected async Task<bool> AddToInboxAsync( |
|||
string messageId, |
|||
string eventName, |
|||
Type eventType, |
|||
byte[] eventBytes) |
|||
{ |
|||
if (AbpDistributedEventBusOptions.Inboxes.Count <= 0) |
|||
{ |
|||
return false; |
|||
} |
|||
|
|||
using (var scope = ServiceScopeFactory.CreateScope()) |
|||
{ |
|||
foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values) |
|||
{ |
|||
if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType)) |
|||
{ |
|||
var eventInbox = (IEventInbox) scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); |
|||
|
|||
if (!messageId.IsNullOrEmpty()) |
|||
{ |
|||
if (await eventInbox.ExistsByMessageIdAsync(messageId)) |
|||
{ |
|||
continue; |
|||
} |
|||
} |
|||
|
|||
await eventInbox.EnqueueAsync( |
|||
new IncomingEventInfo( |
|||
GuidGenerator.Create(), |
|||
messageId, |
|||
eventName, |
|||
eventBytes, |
|||
Clock.Now |
|||
) |
|||
); |
|||
} |
|||
} |
|||
} |
|||
|
|||
return true; |
|||
} |
|||
|
|||
protected abstract byte[] Serialize(object eventData); |
|||
} |
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public interface IEventInbox |
|||
{ |
|||
Task EnqueueAsync(IncomingEventInfo incomingEvent); |
|||
|
|||
Task<List<IncomingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default); |
|||
|
|||
Task MarkAsProcessedAsync(Guid id); |
|||
|
|||
Task<bool> ExistsByMessageIdAsync(string messageId); |
|||
|
|||
Task DeleteOldEventsAsync(); |
|||
} |
|||
} |
|||
@ -0,0 +1,16 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public interface IEventOutbox |
|||
{ |
|||
Task EnqueueAsync(OutgoingEventInfo outgoingEvent); |
|||
|
|||
Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default); |
|||
|
|||
Task DeleteAsync(Guid id); |
|||
} |
|||
} |
|||
@ -0,0 +1,17 @@ |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public interface ISupportsEventBoxes |
|||
{ |
|||
Task PublishFromOutboxAsync( |
|||
OutgoingEventInfo outgoingEvent, |
|||
OutboxConfig outboxConfig |
|||
); |
|||
|
|||
Task ProcessFromInboxAsync( |
|||
IncomingEventInfo incomingEvent, |
|||
InboxConfig inboxConfig |
|||
); |
|||
} |
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
using System; |
|||
using JetBrains.Annotations; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class InboxConfig |
|||
{ |
|||
[NotNull] |
|||
public string Name { get; } |
|||
|
|||
public Type ImplementationType { get; set; } |
|||
|
|||
public Func<Type, bool> EventSelector { get; set; } |
|||
|
|||
public Func<Type, bool> HandlerSelector { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// Used to enable/disable processing incoming events.
|
|||
/// Default: true.
|
|||
/// </summary>
|
|||
public bool IsProcessingEnabled { get; set; } = true; |
|||
|
|||
public InboxConfig([NotNull] string name) |
|||
{ |
|||
Name = Check.NotNullOrWhiteSpace(name, nameof(name)); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,19 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class InboxConfigDictionary : Dictionary<string, InboxConfig> |
|||
{ |
|||
public void Configure(Action<InboxConfig> configAction) |
|||
{ |
|||
Configure("Default", configAction); |
|||
} |
|||
|
|||
public void Configure(string outboxName, Action<InboxConfig> configAction) |
|||
{ |
|||
var outboxConfig = this.GetOrAdd(outboxName, () => new InboxConfig(outboxName)); |
|||
configAction(outboxConfig); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
using System; |
|||
using Volo.Abp.Data; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class IncomingEventInfo : IHasExtraProperties |
|||
{ |
|||
public static int MaxEventNameLength { get; set; } = 256; |
|||
|
|||
public ExtraPropertyDictionary ExtraProperties { get; protected set; } |
|||
|
|||
public Guid Id { get; } |
|||
|
|||
public string MessageId { get; } |
|||
|
|||
public string EventName { get; } |
|||
|
|||
public byte[] EventData { get; } |
|||
|
|||
public DateTime CreationTime { get; } |
|||
|
|||
protected IncomingEventInfo() |
|||
{ |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public IncomingEventInfo( |
|||
Guid id, |
|||
string messageId, |
|||
string eventName, |
|||
byte[] eventData, |
|||
DateTime creationTime) |
|||
{ |
|||
Id = id; |
|||
MessageId = messageId; |
|||
EventName = Check.NotNullOrWhiteSpace(eventName, nameof(eventName), MaxEventNameLength); |
|||
EventData = eventData; |
|||
CreationTime = creationTime; |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
using System; |
|||
using JetBrains.Annotations; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class OutboxConfig |
|||
{ |
|||
[NotNull] |
|||
public string Name { get; } |
|||
|
|||
public Type ImplementationType { get; set; } |
|||
|
|||
public Func<Type, bool> Selector { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// Used to enable/disable sending events from outbox to the message broker.
|
|||
/// Default: true.
|
|||
/// </summary>
|
|||
public bool IsSendingEnabled { get; set; } = true; |
|||
|
|||
public OutboxConfig([NotNull] string name) |
|||
{ |
|||
Name = Check.NotNullOrWhiteSpace(name, nameof(name)); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,19 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class OutboxConfigDictionary : Dictionary<string, OutboxConfig> |
|||
{ |
|||
public void Configure(Action<OutboxConfig> configAction) |
|||
{ |
|||
Configure("Default", configAction); |
|||
} |
|||
|
|||
public void Configure(string outboxName, Action<OutboxConfig> configAction) |
|||
{ |
|||
var outboxConfig = this.GetOrAdd(outboxName, () => new OutboxConfig(outboxName)); |
|||
configAction(outboxConfig); |
|||
} |
|||
} |
|||
} |
|||
@ -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 = Check.NotNullOrWhiteSpace(eventName, nameof(eventName), MaxEventNameLength); |
|||
EventData = eventData; |
|||
CreationTime = creationTime; |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
}} |
|||
@ -0,0 +1,9 @@ |
|||
using MongoDB.Driver; |
|||
|
|||
namespace Volo.Abp.MongoDB.DistributedEvents |
|||
{ |
|||
public interface IHasEventInbox : IAbpMongoDbContext |
|||
{ |
|||
IMongoCollection<IncomingEventRecord> IncomingEvents { get; } |
|||
} |
|||
} |
|||
@ -0,0 +1,9 @@ |
|||
using MongoDB.Driver; |
|||
|
|||
namespace Volo.Abp.MongoDB.DistributedEvents |
|||
{ |
|||
public interface IHasEventOutbox : IAbpMongoDbContext |
|||
{ |
|||
IMongoCollection<OutgoingEventRecord> OutgoingEvents { get; } |
|||
} |
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.MongoDB.DistributedEvents |
|||
{ |
|||
public interface IMongoDbContextEventInbox<TDbContext> : IEventInbox |
|||
where TDbContext : IHasEventInbox |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.MongoDB.DistributedEvents |
|||
{ |
|||
public interface IMongoDbContextEventOutbox<TDbContext> : IEventOutbox |
|||
where TDbContext : IHasEventOutbox |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,66 @@ |
|||
using System; |
|||
using Volo.Abp.Auditing; |
|||
using Volo.Abp.Data; |
|||
using Volo.Abp.Domain.Entities; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
|
|||
namespace Volo.Abp.MongoDB.DistributedEvents |
|||
{ |
|||
public class IncomingEventRecord : |
|||
BasicAggregateRoot<Guid>, |
|||
IHasExtraProperties, |
|||
IHasCreationTime |
|||
{ |
|||
public static int MaxEventNameLength { get; set; } = 256; |
|||
|
|||
public ExtraPropertyDictionary ExtraProperties { get; private set; } |
|||
|
|||
public string MessageId { get; private set; } |
|||
|
|||
public string EventName { get; private set; } |
|||
|
|||
public byte[] EventData { get; private set; } |
|||
|
|||
public DateTime CreationTime { get; private set; } |
|||
|
|||
public bool Processed { get; set; } |
|||
|
|||
public DateTime? ProcessedTime { get; set; } |
|||
|
|||
protected IncomingEventRecord() |
|||
{ |
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public IncomingEventRecord( |
|||
IncomingEventInfo eventInfo) |
|||
: base(eventInfo.Id) |
|||
{ |
|||
MessageId = eventInfo.MessageId; |
|||
EventName = eventInfo.EventName; |
|||
EventData = eventInfo.EventData; |
|||
CreationTime = eventInfo.CreationTime; |
|||
|
|||
ExtraProperties = new ExtraPropertyDictionary(); |
|||
this.SetDefaultsForExtraProperties(); |
|||
} |
|||
|
|||
public IncomingEventInfo ToIncomingEventInfo() |
|||
{ |
|||
return new IncomingEventInfo( |
|||
Id, |
|||
MessageId, |
|||
EventName, |
|||
EventData, |
|||
CreationTime |
|||
); |
|||
} |
|||
|
|||
public void MarkAsProcessed(DateTime processedTime) |
|||
{ |
|||
Processed = true; |
|||
ProcessedTime = processedTime; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,112 @@ |
|||
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.Boxes; |
|||
using Volo.Abp.EventBus.Distributed; |
|||
using Volo.Abp.Timing; |
|||
using Volo.Abp.Uow; |
|||
|
|||
namespace Volo.Abp.MongoDB.DistributedEvents |
|||
{ |
|||
public class MongoDbContextEventInbox<TMongoDbContext> : IMongoDbContextEventInbox<TMongoDbContext> |
|||
where TMongoDbContext : IHasEventInbox |
|||
{ |
|||
protected IMongoDbContextProvider<TMongoDbContext> DbContextProvider { get; } |
|||
protected AbpEventBusBoxesOptions EventBusBoxesOptions { get; } |
|||
protected IClock Clock { get; } |
|||
|
|||
public MongoDbContextEventInbox( |
|||
IMongoDbContextProvider<TMongoDbContext> dbContextProvider, |
|||
IClock clock, |
|||
IOptions<AbpEventBusBoxesOptions> eventBusBoxesOptions) |
|||
{ |
|||
DbContextProvider = dbContextProvider; |
|||
Clock = clock; |
|||
EventBusBoxesOptions = eventBusBoxesOptions.Value; |
|||
} |
|||
|
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task EnqueueAsync(IncomingEventInfo incomingEvent) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
if (dbContext.SessionHandle != null) |
|||
{ |
|||
await dbContext.IncomingEvents.InsertOneAsync( |
|||
dbContext.SessionHandle, |
|||
new IncomingEventRecord(incomingEvent) |
|||
); |
|||
} |
|||
else |
|||
{ |
|||
await dbContext.IncomingEvents.InsertOneAsync( |
|||
new IncomingEventRecord(incomingEvent) |
|||
); |
|||
} |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task<List<IncomingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(cancellationToken); |
|||
|
|||
var outgoingEventRecords = await dbContext |
|||
.IncomingEvents |
|||
.AsQueryable() |
|||
.Where(x => !x.Processed) |
|||
.OrderBy(x => x.CreationTime) |
|||
.Take(maxCount) |
|||
.ToListAsync(cancellationToken: cancellationToken); |
|||
|
|||
return outgoingEventRecords |
|||
.Select(x => x.ToIncomingEventInfo()) |
|||
.ToList(); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task MarkAsProcessedAsync(Guid id) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
|
|||
var filter = Builders<IncomingEventRecord>.Filter.Eq(x => x.Id, id); |
|||
var update = Builders<IncomingEventRecord>.Update.Set(x => x.Processed, true).Set(x => x.ProcessedTime, Clock.Now); |
|||
|
|||
if (dbContext.SessionHandle != null) |
|||
{ |
|||
await dbContext.IncomingEvents.UpdateOneAsync(dbContext.SessionHandle, filter, update); |
|||
} |
|||
else |
|||
{ |
|||
await dbContext.IncomingEvents.UpdateOneAsync(filter, update); |
|||
} |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task<bool> ExistsByMessageIdAsync(string messageId) |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
return await dbContext.IncomingEvents.AsQueryable().AnyAsync(x => x.MessageId == messageId); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task DeleteOldEventsAsync() |
|||
{ |
|||
var dbContext = await DbContextProvider.GetDbContextAsync(); |
|||
var timeToKeepEvents = Clock.Now - EventBusBoxesOptions.WaitTimeToDeleteProcessedInboxEvents; |
|||
|
|||
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); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +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<TMongoDbContext> : IMongoDbContextEventOutbox<TMongoDbContext> |
|||
where TMongoDbContext : IHasEventOutbox |
|||
{ |
|||
protected IMongoDbContextProvider<TMongoDbContext> MongoDbContextProvider { get; } |
|||
|
|||
public MongoDbContextEventOutbox(IMongoDbContextProvider<TMongoDbContext> mongoDbContextProvider) |
|||
{ |
|||
MongoDbContextProvider = mongoDbContextProvider; |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task EnqueueAsync(OutgoingEventInfo outgoingEvent) |
|||
{ |
|||
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) |
|||
); |
|||
} |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default) |
|||
{ |
|||
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(); |
|||
} |
|||
|
|||
[UnitOfWork] |
|||
public virtual async Task DeleteAsync(Guid id) |
|||
{ |
|||
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)); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue