Eliminate duplicate incoming events

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<List<IncomingEventInfo>> 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<bool> ExistsByMessageIdAsync(string messageId)
+        {
+            var dbContext = await GetDbContextAsync();
+            return await dbContext.IncomingEvents.AnyAsync(x => x.MessageId == messageId);
+        }
+
+        private async Task<IHasEventInbox> GetDbContextAsync()
+        {
+            return (IHasEventInbox)await DbContextProvider.GetDbContextAsync();
+        }
+
+        public Task DeleteOldEventsAsync()
+        {
+            throw new NotImplementedException();
+        }
     }
 }

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);
             });
         }
     }

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

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
             );

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(

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

@@ -113,6 +113,7 @@ namespace Volo.Abp.EventBus.Distributed
         }
 
         protected async Task<bool> 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

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventInbox.cs

@@ -11,5 +11,9 @@ namespace Volo.Abp.EventBus.Distributed
         Task<List<IncomingEventInfo>> GetWaitingEventsAsync(int maxCount);
         
         Task MarkAsProcessedAsync(Guid id);
+        
+        Task<bool> ExistsByMessageIdAsync(string messageId);
+
+        Task DeleteOldEventsAsync();
     }
 }

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;

test/DistEvents/DistDemoApp/Migrations/20210910111009_Added_Inbox_New_Cols.Designer.cs

@@ -0,0 +1,162 @@
+// <auto-generated />
+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<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    b.Property<string>("ConcurrencyStamp")
+                        .IsConcurrencyToken()
+                        .HasMaxLength(40)
+                        .HasColumnType("nvarchar(40)")
+                        .HasColumnName("ConcurrencyStamp");
+
+                    b.Property<DateTime>("CreationTime")
+                        .HasColumnType("datetime2")
+                        .HasColumnName("CreationTime");
+
+                    b.Property<Guid?>("CreatorId")
+                        .HasColumnType("uniqueidentifier")
+                        .HasColumnName("CreatorId");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.Property<string>("Text")
+                        .IsRequired()
+                        .HasMaxLength(128)
+                        .HasColumnType("nvarchar(128)");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("TodoItems");
+                });
+
+            modelBuilder.Entity("DistDemoApp.TodoSummary", b =>
+                {
+                    b.Property<int>("Id")
+                        .ValueGeneratedOnAdd()
+                        .HasColumnType("int")
+                        .HasAnnotation("SqlServer:ValueGenerationStrategy", SqlServerValueGenerationStrategy.IdentityColumn);
+
+                    b.Property<string>("ConcurrencyStamp")
+                        .IsConcurrencyToken()
+                        .HasMaxLength(40)
+                        .HasColumnType("nvarchar(40)")
+                        .HasColumnName("ConcurrencyStamp");
+
+                    b.Property<byte>("Day")
+                        .HasColumnType("tinyint");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.Property<byte>("Month")
+                        .HasColumnType("tinyint");
+
+                    b.Property<int>("TotalCount")
+                        .HasColumnType("int");
+
+                    b.Property<int>("Year")
+                        .HasColumnType("int");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("TodoSummaries");
+                });
+
+            modelBuilder.Entity("Volo.Abp.EntityFrameworkCore.DistributedEvents.IncomingEventRecord", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    b.Property<DateTime>("CreationTime")
+                        .HasColumnType("datetime2")
+                        .HasColumnName("CreationTime");
+
+                    b.Property<byte[]>("EventData")
+                        .IsRequired()
+                        .HasColumnType("varbinary(max)");
+
+                    b.Property<string>("EventName")
+                        .IsRequired()
+                        .HasMaxLength(256)
+                        .HasColumnType("nvarchar(256)");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.Property<string>("MessageId")
+                        .HasColumnType("nvarchar(450)");
+
+                    b.Property<bool>("Processed")
+                        .HasColumnType("bit");
+
+                    b.Property<DateTime?>("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<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    b.Property<DateTime>("CreationTime")
+                        .HasColumnType("datetime2")
+                        .HasColumnName("CreationTime");
+
+                    b.Property<byte[]>("EventData")
+                        .IsRequired()
+                        .HasColumnType("varbinary(max)");
+
+                    b.Property<string>("EventName")
+                        .IsRequired()
+                        .HasMaxLength(256)
+                        .HasColumnType("nvarchar(256)");
+
+                    b.Property<string>("ExtraProperties")
+                        .HasColumnType("nvarchar(max)")
+                        .HasColumnName("ExtraProperties");
+
+                    b.HasKey("Id");
+
+                    b.ToTable("AbpEventOutbox");
+                });
+#pragma warning restore 612, 618
+        }
+    }
+}

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<string>(
+                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");
+        }
+    }
+}

test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs

@@ -110,6 +110,9 @@ namespace DistDemoApp.Migrations
                         .HasColumnType("nvarchar(max)")
                         .HasColumnName("ExtraProperties");
 
+                    b.Property<string>("MessageId")
+                        .HasColumnType("nvarchar(450)");
+
                     b.Property<bool>("Processed")
                         .HasColumnType("bit");
 
@@ -118,6 +121,10 @@ namespace DistDemoApp.Migrations
 
                     b.HasKey("Id");
 
+                    b.HasIndex("MessageId");
+
+                    b.HasIndex("Processed", "CreationTime");
+
                     b.ToTable("AbpEventInbox");
                 });