From 51ce477165843aa975034a6c47cd24b1351ef4bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Halil=20=C4=B0brahim=20Kalkan?= Date: Fri, 10 Sep 2021 14:30:10 +0300 Subject: [PATCH] Eliminate duplicate incoming events --- .../DistributedEvents/DbContextEventInbox.cs | 27 ++- ...entInboxDbContextModelBuilderExtensions.cs | 3 + .../DistributedEvents/IncomingEventRecord.cs | 4 + .../Kafka/KafkaDistributedEventBus.cs | 13 +- .../RabbitMq/RabbitMqDistributedEventBus.cs | 10 +- .../Distributed/DistributedEventBusBase.cs | 12 +- .../Abp/EventBus/Distributed/IEventInbox.cs | 4 + .../EventBus/Distributed/IncomingEventInfo.cs | 4 + ...910111009_Added_Inbox_New_Cols.Designer.cs | 162 ++++++++++++++++++ .../20210910111009_Added_Inbox_New_Cols.cs | 41 +++++ .../Migrations/TodoDbContextModelSnapshot.cs | 7 + 11 files changed, 276 insertions(+), 11 deletions(-) create mode 100644 test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.Designer.cs create mode 100644 test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.cs diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs index df45d50ae5..f5f1d7010e 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs +++ b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs @@ -26,7 +26,8 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents [UnitOfWork] public virtual async Task EnqueueAsync(IncomingEventInfo incomingEvent) { - var dbContext = (IHasEventInbox) await DbContextProvider.GetDbContextAsync(); + var dbContext = await GetDbContextAsync(); + dbContext.IncomingEvents.Add( new IncomingEventRecord(incomingEvent) ); @@ -35,8 +36,8 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents [UnitOfWork] public virtual async Task> GetWaitingEventsAsync(int maxCount) { - var dbContext = (IHasEventInbox) await DbContextProvider.GetDbContextAsync(); - + var dbContext = await GetDbContextAsync(); + var outgoingEventRecords = await dbContext .IncomingEvents .AsNoTracking() @@ -50,15 +51,33 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents .ToList(); } + [UnitOfWork] public async Task MarkAsProcessedAsync(Guid id) { //TODO: Optimize? - var dbContext = (IHasEventInbox) await DbContextProvider.GetDbContextAsync(); + var dbContext = await GetDbContextAsync(); var incomingEvent = await dbContext.IncomingEvents.FindAsync(id); if (incomingEvent != null) { incomingEvent.MarkAsProcessed(Clock.Now); } } + + [UnitOfWork] + public async Task ExistsByMessageIdAsync(string messageId) + { + var dbContext = await GetDbContextAsync(); + return await dbContext.IncomingEvents.AnyAsync(x => x.MessageId == messageId); + } + + private async Task GetDbContextAsync() + { + return (IHasEventInbox)await DbContextProvider.GetDbContextAsync(); + } + + public Task DeleteOldEventsAsync() + { + throw new NotImplementedException(); + } } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/EventInboxDbContextModelBuilderExtensions.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/EventInboxDbContextModelBuilderExtensions.cs index d3e3a6cc26..29c9f0042c 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/EventInboxDbContextModelBuilderExtensions.cs +++ b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/EventInboxDbContextModelBuilderExtensions.cs @@ -15,6 +15,9 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents 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); }); } } diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs index 62c2781b48..a1f971bbaa 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs +++ b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs @@ -15,6 +15,8 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents public ExtraPropertyDictionary ExtraProperties { get; private set; } + public string MessageId { get; private set; } + public string EventName { get; private set; } public byte[] EventData { get; private set; } @@ -35,6 +37,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents IncomingEventInfo eventInfo) : base(eventInfo.Id) { + MessageId = eventInfo.MessageId; EventName = eventInfo.EventName; EventData = eventInfo.EventData; CreationTime = eventInfo.CreationTime; @@ -47,6 +50,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents { return new IncomingEventInfo( Id, + MessageId, EventName, EventData, CreationTime diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs index e587dc9749..3ffe3bb880 100644 --- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs @@ -86,8 +86,15 @@ namespace Volo.Abp.EventBus.Kafka { return; } + + string messageId = null; + + if (message.Headers.TryGetLastBytes("messageId", out var messageIdBytes)) + { + messageId = System.Text.Encoding.UTF8.GetString(messageIdBytes); + } - if (await AddToInboxAsync(eventName, eventType, message.Value)) + if (await AddToInboxAsync(messageId, eventName, eventType, message.Value)) { return; } @@ -183,7 +190,7 @@ namespace Volo.Abp.EventBus.Kafka eventData, new Headers { - { "messageId", Serializer.Serialize(Guid.NewGuid()) } + { "messageId", System.Text.Encoding.UTF8.GetBytes(Guid.NewGuid().ToString("N")) } }, null ); @@ -205,7 +212,7 @@ namespace Volo.Abp.EventBus.Kafka eventData, new Headers { - { "messageId", Serializer.Serialize(eventId) } + { "messageId", System.Text.Encoding.UTF8.GetBytes(eventId.ToString("N")) } }, null ); diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index eb171f7ffb..deb20b5ff3 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -106,8 +106,8 @@ namespace Volo.Abp.EventBus.RabbitMq } var eventBytes = ea.Body.ToArray(); - - if (await AddToInboxAsync(eventName, eventType, eventBytes)) + + if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes)) { return; } @@ -268,9 +268,13 @@ namespace Volo.Abp.EventBus.RabbitMq { properties = channel.CreateBasicProperties(); properties.DeliveryMode = RabbitMqConsts.DeliveryModes.Persistent; - properties.MessageId = (eventId ?? GuidGenerator.Create()).ToString("N"); } + if (properties.MessageId.IsNullOrEmpty()) + { + properties.MessageId = (eventId ?? GuidGenerator.Create()).ToString("N"); + } + SetEventMessageHeaders(properties, headersArguments); channel.BasicPublish( diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs index 8de449058b..1fdaf0d26f 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs @@ -113,6 +113,7 @@ namespace Volo.Abp.EventBus.Distributed } protected async Task AddToInboxAsync( + string messageId, string eventName, Type eventType, byte[] eventBytes) @@ -129,10 +130,19 @@ namespace Volo.Abp.EventBus.Distributed if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType)) { var eventInbox = (IEventInbox) scope.ServiceProvider.GetRequiredService(inboxConfig.ImplementationType); - //TODO: Check if event was received before!! + + if (!messageId.IsNullOrEmpty()) + { + if (await eventInbox.ExistsByMessageIdAsync(messageId)) + { + continue; + } + } + await eventInbox.EnqueueAsync( new IncomingEventInfo( GuidGenerator.Create(), + messageId, eventName, eventBytes, Clock.Now diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs index 137a410afa..bee802a126 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs @@ -11,5 +11,9 @@ namespace Volo.Abp.EventBus.Distributed Task> GetWaitingEventsAsync(int maxCount); Task MarkAsProcessedAsync(Guid id); + + Task ExistsByMessageIdAsync(string messageId); + + Task DeleteOldEventsAsync(); } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs index 7cd628cf91..571c62602b 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs @@ -11,6 +11,8 @@ namespace Volo.Abp.EventBus.Distributed public Guid Id { get; } + public string MessageId { get; } + public string EventName { get; } public byte[] EventData { get; } @@ -25,11 +27,13 @@ namespace Volo.Abp.EventBus.Distributed 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; diff --git a/test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.Designer.cs b/test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.Designer.cs new file mode 100644 index 0000000000..7b552dfcfc --- /dev/null +++ b/test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.Designer.cs @@ -0,0 +1,162 @@ +// +using System; +using DistDemoApp; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Volo.Abp.EntityFrameworkCore; + +namespace DistDemoApp.Migrations +{ + [DbContext(typeof(TodoDbContext))] + [Migration("20210910111009_Added_Inbox_New_Cols")] + partial class Added_Inbox_New_Cols + { + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("_Abp_DatabaseProvider", EfCoreDatabaseProvider.SqlServer) + .HasAnnotation("Relational:MaxIdentifierLength", 128) + .HasAnnotation("ProductVersion", "5.0.9") + .HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.IdentityColumn); + + modelBuilder.Entity("DistDemoApp.TodoItem", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("ConcurrencyStamp") + .IsConcurrencyToken() + .HasMaxLength(40) + .HasColumnType("nvarchar(40)") + .HasColumnName("ConcurrencyStamp"); + + b.Property("CreationTime") + .HasColumnType("datetime2") + .HasColumnName("CreationTime"); + + b.Property("CreatorId") + .HasColumnType("uniqueidentifier") + .HasColumnName("CreatorId"); + + b.Property("ExtraProperties") + .HasColumnType("nvarchar(max)") + .HasColumnName("ExtraProperties"); + + b.Property("Text") + .IsRequired() + .HasMaxLength(128) + .HasColumnType("nvarchar(128)"); + + b.HasKey("Id"); + + b.ToTable("TodoItems"); + }); + + modelBuilder.Entity("DistDemoApp.TodoSummary", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("int") + .HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.IdentityColumn); + + b.Property("ConcurrencyStamp") + .IsConcurrencyToken() + .HasMaxLength(40) + .HasColumnType("nvarchar(40)") + .HasColumnName("ConcurrencyStamp"); + + b.Property("Day") + .HasColumnType("tinyint"); + + b.Property("ExtraProperties") + .HasColumnType("nvarchar(max)") + .HasColumnName("ExtraProperties"); + + b.Property("Month") + .HasColumnType("tinyint"); + + b.Property("TotalCount") + .HasColumnType("int"); + + b.Property("Year") + .HasColumnType("int"); + + b.HasKey("Id"); + + b.ToTable("TodoSummaries"); + }); + + modelBuilder.Entity("Volo.Abp.EntityFrameworkCore.DistributedEvents.IncomingEventRecord", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("CreationTime") + .HasColumnType("datetime2") + .HasColumnName("CreationTime"); + + b.Property("EventData") + .IsRequired() + .HasColumnType("varbinary(max)"); + + b.Property("EventName") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("nvarchar(256)"); + + b.Property("ExtraProperties") + .HasColumnType("nvarchar(max)") + .HasColumnName("ExtraProperties"); + + b.Property("MessageId") + .HasColumnType("nvarchar(450)"); + + b.Property("Processed") + .HasColumnType("bit"); + + b.Property("ProcessedTime") + .HasColumnType("datetime2"); + + b.HasKey("Id"); + + b.HasIndex("MessageId"); + + b.HasIndex("Processed", "CreationTime"); + + b.ToTable("AbpEventInbox"); + }); + + modelBuilder.Entity("Volo.Abp.EntityFrameworkCore.DistributedEvents.OutgoingEventRecord", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("CreationTime") + .HasColumnType("datetime2") + .HasColumnName("CreationTime"); + + b.Property("EventData") + .IsRequired() + .HasColumnType("varbinary(max)"); + + b.Property("EventName") + .IsRequired() + .HasMaxLength(256) + .HasColumnType("nvarchar(256)"); + + b.Property("ExtraProperties") + .HasColumnType("nvarchar(max)") + .HasColumnName("ExtraProperties"); + + b.HasKey("Id"); + + b.ToTable("AbpEventOutbox"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.cs b/test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.cs new file mode 100644 index 0000000000..24f65ff23c --- /dev/null +++ b/test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.cs @@ -0,0 +1,41 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +namespace DistDemoApp.Migrations +{ + public partial class Added_Inbox_New_Cols : Migration + { + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "MessageId", + table: "AbpEventInbox", + type: "nvarchar(450)", + nullable: true); + + migrationBuilder.CreateIndex( + name: "IX_AbpEventInbox_MessageId", + table: "AbpEventInbox", + column: "MessageId"); + + migrationBuilder.CreateIndex( + name: "IX_AbpEventInbox_Processed_CreationTime", + table: "AbpEventInbox", + columns: new[] { "Processed", "CreationTime" }); + } + + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "IX_AbpEventInbox_MessageId", + table: "AbpEventInbox"); + + migrationBuilder.DropIndex( + name: "IX_AbpEventInbox_Processed_CreationTime", + table: "AbpEventInbox"); + + migrationBuilder.DropColumn( + name: "MessageId", + table: "AbpEventInbox"); + } + } +} diff --git a/test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs b/test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs index 87cf72bfd0..57e8c14442 100644 --- a/test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs +++ b/test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs @@ -110,6 +110,9 @@ namespace DistDemoApp.Migrations .HasColumnType("nvarchar(max)") .HasColumnName("ExtraProperties"); + b.Property("MessageId") + .HasColumnType("nvarchar(450)"); + b.Property("Processed") .HasColumnType("bit"); @@ -118,6 +121,10 @@ namespace DistDemoApp.Migrations b.HasKey("Id"); + b.HasIndex("MessageId"); + + b.HasIndex("Processed", "CreationTime"); + b.ToTable("AbpEventInbox"); });