Save incoming events to inbox

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/AbpEntityFrameworkCoreModule.cs

@@ -28,6 +28,7 @@ namespace Volo.Abp.EntityFrameworkCore
 
             context.Services.TryAddTransient(typeof(IDbContextProvider<>), typeof(UnitOfWorkDbContextProvider<>));
             context.Services.AddTransient(typeof(IDbContextEventOutbox<>), typeof(DbContextEventOutbox<>));
+            context.Services.AddTransient(typeof(IDbContextEventInbox<>), typeof(DbContextEventInbox<>));
         }
     }
 }

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs

@@ -0,0 +1,47 @@
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.Uow;
+
+namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
+{
+    public class DbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext> 
+        where TDbContext : IHasEventInbox
+    {
+        protected IDbContextProvider<TDbContext> DbContextProvider { get; }
+
+        public DbContextEventInbox(
+            IDbContextProvider<TDbContext> dbContextProvider)
+        {
+            DbContextProvider = dbContextProvider;
+        }
+
+        [UnitOfWork]
+        public virtual async Task EnqueueAsync(IncomingEventInfo incomingEvent)
+        {
+            var dbContext = (IHasEventInbox) await DbContextProvider.GetDbContextAsync();
+            dbContext.IncomingEvents.Add(
+                new IncomingEventRecord(incomingEvent)
+            );
+        }
+
+        [UnitOfWork]
+        public virtual async Task<List<IncomingEventInfo>> GetWaitingEventsAsync(int maxCount)
+        {
+            var dbContext = (IHasEventInbox) await DbContextProvider.GetDbContextAsync();
+            
+            var outgoingEventRecords = await dbContext
+                .IncomingEvents
+                .AsNoTracking()
+                .OrderBy(x => x.CreationTime)
+                .Take(maxCount)
+                .ToListAsync();
+            
+            return outgoingEventRecords
+                .Select(x => x.ToIncomingEventInfo())
+                .ToList();
+        }
+    }
+}

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs

@@ -23,7 +23,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
         public virtual async Task EnqueueAsync(OutgoingEventInfo outgoingEvent)
         {
             var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync();
-            dbContext.OutgoingEventRecords.Add(
+            dbContext.OutgoingEvents.Add(
                 new OutgoingEventRecord(outgoingEvent)
             );
         }
@@ -34,7 +34,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
             var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync();
             
             var outgoingEventRecords = await dbContext
-                .OutgoingEventRecords
+                .OutgoingEvents
                 .AsNoTracking()
                 .OrderBy(x => x.CreationTime)
                 .Take(maxCount)
@@ -49,7 +49,7 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
         public virtual async Task DeleteAsync(Guid id)
         {
             var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync();
-            var outgoingEvent = await dbContext.OutgoingEventRecords.FindAsync(id);
+            var outgoingEvent = await dbContext.OutgoingEvents.FindAsync(id);
             if (outgoingEvent != null)
             {
                 dbContext.Remove(outgoingEvent);

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/EfCoreInboxConfigExtensions.cs

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

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/EventInboxDbContextModelBuilderExtensions.cs

@@ -0,0 +1,21 @@
+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();
+            });
+        }
+    }
+}

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IDbContextEventInbox.cs

@@ -0,0 +1,10 @@
+using Volo.Abp.EventBus.Distributed;
+
+namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
+{
+    public interface IDbContextEventInbox<TDbContext> : IEventInbox
+        where TDbContext : IHasEventInbox
+    {
+        
+    }
+}

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IHasEventInbox.cs

@@ -0,0 +1,9 @@
+using Microsoft.EntityFrameworkCore;
+
+namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
+{
+    public interface IHasEventInbox : IEfCoreDbContext
+    {
+        DbSet<IncomingEventRecord> IncomingEvents { get; set; }
+    }
+}

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IHasEventOutbox.cs

@@ -4,6 +4,6 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
 {
     public interface IHasEventOutbox : IEfCoreDbContext
     {
-        DbSet<OutgoingEventRecord> OutgoingEventRecords { get; set; }
+        DbSet<OutgoingEventRecord> OutgoingEvents { get; set; }
     }
 }

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs

@@ -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 IncomingEventRecord : 
+        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 IncomingEventRecord()
+        {
+            ExtraProperties = new ExtraPropertyDictionary();
+            this.SetDefaultsForExtraProperties();
+        }
+
+        public IncomingEventRecord(
+            IncomingEventInfo eventInfo)
+            : base(eventInfo.Id)
+        {
+            EventName = eventInfo.EventName;
+            EventData = eventInfo.EventData;
+            CreationTime = eventInfo.CreationTime;
+
+            ExtraProperties = new ExtraPropertyDictionary();
+            this.SetDefaultsForExtraProperties();
+        }
+
+        public IncomingEventInfo ToIncomingEventInfo()
+        {
+            return new IncomingEventInfo(
+                Id,
+                EventName,
+                EventData,
+                CreationTime
+            );
+        }
+    }
+}

framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs

@@ -86,6 +86,11 @@ namespace Volo.Abp.EventBus.Kafka
             {
                 return;
             }
+            
+            if (await AddToInboxAsync(eventName, eventType, message.Value))
+            {
+                return;
+            }
 
             var eventData = Serializer.Deserialize(message.Value, eventType);
 

framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs

@@ -105,7 +105,14 @@ namespace Volo.Abp.EventBus.RabbitMq
                 return;
             }
 
-            var eventData = Serializer.Deserialize(ea.Body.ToArray(), eventType);
+            var eventBytes = ea.Body.ToArray();
+
+            if (await AddToInboxAsync(eventName, eventType, eventBytes))
+            {
+                return;
+            }
+            
+            var eventData = Serializer.Deserialize(eventBytes, eventType);
 
             await TriggerHandlersAsync(eventType, eventData, errorContext =>
             {

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

@@ -6,12 +6,15 @@ namespace Volo.Abp.EventBus.Distributed
     {
         public ITypeList<IEventHandler> Handlers { get; }
         
-        public OutboxConfigList Outboxes { get; }
+        public OutboxConfigDictionary Outboxes { get; }
+        
+        public InboxConfigDictionary Inboxes { get; }
 
         public AbpDistributedEventBusOptions()
         {
             Handlers = new TypeList<IEventHandler>();
-            Outboxes = new OutboxConfigList();
+            Outboxes = new OutboxConfigDictionary();
+            Inboxes = new InboxConfigDictionary();
         }
     }
 }

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

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
@@ -17,14 +18,14 @@ namespace Volo.Abp.EventBus.Distributed
 
         protected DistributedEventBusBase(
             IServiceScopeFactory serviceScopeFactory,
-            ICurrentTenant currentTenant, 
+            ICurrentTenant currentTenant,
             IUnitOfWorkManager unitOfWorkManager,
             IEventErrorHandler errorHandler,
             IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
             IGuidGenerator guidGenerator,
             IClock clock
-            ) : base(
-            serviceScopeFactory, 
+        ) : base(
+            serviceScopeFactory,
             currentTenant,
             unitOfWorkManager,
             errorHandler)
@@ -67,7 +68,7 @@ namespace Volo.Abp.EventBus.Distributed
                 );
                 return;
             }
-            
+
             if (useOutbox)
             {
                 if (await AddToOutboxAsync(eventType, eventData))
@@ -106,10 +107,42 @@ namespace Volo.Abp.EventBus.Distributed
                     return true;
                 }
             }
-            
+
             return false;
         }
 
+        protected async Task<bool> AddToInboxAsync(
+            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);
+                        await eventInbox.EnqueueAsync(
+                            new IncomingEventInfo(
+                                GuidGenerator.Create(),
+                                eventName,
+                                eventBytes,
+                                Clock.Now
+                            )
+                        );
+                    }
+                }
+            }
+
+            return true;
+        }
+
         protected abstract byte[] Serialize(object eventData);
     }
 }

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

@@ -0,0 +1,12 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Volo.Abp.EventBus.Distributed
+{
+    public interface IEventInbox
+    {
+        Task EnqueueAsync(IncomingEventInfo incomingEvent);
+        
+        Task<List<IncomingEventInfo>> GetWaitingEventsAsync(int maxCount);
+    }
+}

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

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

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

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

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

@@ -0,0 +1,40 @@
+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 EventName { get; }
+        
+        public byte[] EventData { get; }
+        
+        public DateTime CreationTime { get; }
+        
+        protected IncomingEventInfo()
+        {
+            ExtraProperties = new ExtraPropertyDictionary();
+            this.SetDefaultsForExtraProperties();
+        }
+
+        public IncomingEventInfo(
+            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();
+        }
+    }
+}

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxConfigList.cs → framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxConfigDictionary.cs

@@ -3,8 +3,13 @@ using System.Collections.Generic;
 
 namespace Volo.Abp.EventBus.Distributed
 {
-    public class OutboxConfigList : Dictionary<string, OutboxConfig>
+    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));

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

@@ -30,7 +30,7 @@ namespace Volo.Abp.EventBus.Distributed
             DateTime creationTime)
         {
             Id = id;
-            EventName = eventName;
+            EventName = Check.NotNullOrWhiteSpace(eventName, nameof(eventName), MaxEventNameLength);
             EventData = eventData;
             CreationTime = creationTime;
             ExtraProperties = new ExtraPropertyDictionary();

test/DistEvents/DistDemoApp/DistDemoAppModule.cs

@@ -46,7 +46,12 @@ namespace DistDemoApp
             
             Configure<AbpDistributedEventBusOptions>(options =>
             {
-                options.Outboxes.Configure("Default", config =>
+                options.Outboxes.Configure(config =>
+                {
+                    config.UseDbContext<TodoDbContext>();
+                });
+                
+                options.Inboxes.Configure(config =>
                 {
                     config.UseDbContext<TodoDbContext>();
                 });

test/DistEvents/DistDemoApp/Migrations/20210909113934_Added_Inbox.Designer.cs

@@ -0,0 +1,149 @@
+// <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("20210909113934_Added_Inbox")]
+    partial class Added_Inbox
+    {
+        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.HasKey("Id");
+
+                    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/20210909113934_Added_Inbox.cs

@@ -0,0 +1,32 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+namespace DistDemoApp.Migrations
+{
+    public partial class Added_Inbox : Migration
+    {
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.CreateTable(
+                name: "AbpEventInbox",
+                columns: table => new
+                {
+                    Id = table.Column<Guid>(type: "uniqueidentifier", nullable: false),
+                    ExtraProperties = table.Column<string>(type: "nvarchar(max)", nullable: true),
+                    EventName = table.Column<string>(type: "nvarchar(256)", maxLength: 256, nullable: false),
+                    EventData = table.Column<byte[]>(type: "varbinary(max)", nullable: false),
+                    CreationTime = table.Column<DateTime>(type: "datetime2", nullable: false)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_AbpEventInbox", x => x.Id);
+                });
+        }
+
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "AbpEventInbox");
+        }
+    }
+}

test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs

@@ -88,6 +88,33 @@ namespace DistDemoApp.Migrations
                     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.HasKey("Id");
+
+                    b.ToTable("AbpEventInbox");
+                });
+
             modelBuilder.Entity("Volo.Abp.EntityFrameworkCore.DistributedEvents.OutgoingEventRecord", b =>
                 {
                     b.Property<Guid>("Id")

test/DistEvents/DistDemoApp/TodoDbContext.cs

@@ -5,11 +5,12 @@ using Volo.Abp.EntityFrameworkCore.DistributedEvents;
 
 namespace DistDemoApp
 {
-    public class TodoDbContext : AbpDbContext<TodoDbContext>, IHasEventOutbox
+    public class TodoDbContext : AbpDbContext<TodoDbContext>, IHasEventOutbox, IHasEventInbox
     {
         public DbSet<TodoItem> TodoItems { get; set; }
         public DbSet<TodoSummary> TodoSummaries { get; set; }
-        public DbSet<OutgoingEventRecord> OutgoingEventRecords { get; set; }
+        public DbSet<OutgoingEventRecord> OutgoingEvents { get; set; }
+        public DbSet<IncomingEventRecord> IncomingEvents { get; set; }
 
         public TodoDbContext(DbContextOptions<TodoDbContext> options)
             : base(options)
@@ -22,6 +23,7 @@ namespace DistDemoApp
             base.OnModelCreating(modelBuilder);
             
             modelBuilder.ConfigureEventOutbox();
+            modelBuilder.ConfigureEventInbox();
 
             modelBuilder.Entity<TodoItem>(b =>
             {