mirror of https://github.com/abpframework/abp.git
Halil İbrahim Kalkan
5 years ago
committed by
GitHub
GitHub
249 changed files with 4079 additions and 601 deletions
@ -1,5 +1,14 @@ |
|||||
# ABP Framework v4.x to v5.0 Migration Guide |
# ABP Framework v4.x to v5.0 Migration Guide |
||||
|
|
||||
|
## MongoDB |
||||
|
|
||||
|
ABP Framework will serialize the datetime based on [AbpClockOptions](https://docs.abp.io/en/abp/latest/Timing#clock-options) start from 5.0, before `DateTime` values in MongoDB are [always saved as UTC](https://mongodb.github.io/mongo-csharp-driver/2.13/reference/bson/mapping/#datetime-serialization-options). |
||||
|
|
||||
|
You can disable this behavior by configure `AbpMongoDbOptions`. |
||||
|
```cs |
||||
|
services.Configure<AbpMongoDbOptions>(x => x.UseAbpClockHandleDateTime = false); |
||||
|
``` |
||||
|
|
||||
## Angular UI |
## Angular UI |
||||
|
|
||||
See the [Angular UI Migration Guide](Abp-5_0-Angular.md). |
See the [Angular UI Migration Guide](Abp-5_0-Angular.md). |
||||
|
|||||
@ -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,43 @@ |
|||||
|
using System; |
||||
|
using MongoDB.Bson; |
||||
|
using MongoDB.Bson.Serialization; |
||||
|
using MongoDB.Bson.Serialization.Serializers; |
||||
|
|
||||
|
namespace Volo.Abp.MongoDB |
||||
|
{ |
||||
|
public class AbpMongoDbDateTimeSerializer : DateTimeSerializer |
||||
|
{ |
||||
|
protected DateTimeKind DateTimeKind { get; set; } |
||||
|
protected bool DisableDateTimeNormalization{ get; set; } |
||||
|
|
||||
|
public AbpMongoDbDateTimeSerializer(DateTimeKind dateTimeKind , bool disableDateTimeNormalization) |
||||
|
{ |
||||
|
DateTimeKind = dateTimeKind; |
||||
|
DisableDateTimeNormalization = disableDateTimeNormalization; |
||||
|
} |
||||
|
|
||||
|
public override void Serialize(BsonSerializationContext context, BsonSerializationArgs args, DateTime value) |
||||
|
{ |
||||
|
context.Writer.WriteDateTime(DisableDateTimeNormalization |
||||
|
? ToMillisecondsSinceEpoch(value) |
||||
|
: ToMillisecondsSinceEpoch(DateTime.SpecifyKind(value, DateTimeKind))); |
||||
|
} |
||||
|
|
||||
|
public override DateTime Deserialize(BsonDeserializationContext context, BsonDeserializationArgs args) |
||||
|
{ |
||||
|
var dateTime = new BsonDateTime(context.Reader.ReadDateTime()).ToUniversalTime(); |
||||
|
return DateTime.SpecifyKind(dateTime, DisableDateTimeNormalization ? DateTimeKind.Unspecified : DateTimeKind); |
||||
|
} |
||||
|
|
||||
|
private static long ToMillisecondsSinceEpoch(DateTime dateTime) |
||||
|
{ |
||||
|
return (dateTime - BsonConstants.UnixEpoch).Ticks / 10000L; |
||||
|
} |
||||
|
|
||||
|
// For unit testing.
|
||||
|
internal void SetDateTimeKind(DateTimeKind dateTimeKind) |
||||
|
{ |
||||
|
DateTimeKind = dateTimeKind; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue