save to outbox

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

@@ -201,7 +201,7 @@ namespace Volo.Abp.EntityFrameworkCore
             foreach (var distributedEvent in changeReport.DistributedEvents)
             {
                 UnitOfWorkManager.Current?.AddOrReplaceDistributedEvent(
-                    new UnitOfWorkEventRecord(distributedEvent.EventData.GetType(), distributedEvent.EventData, distributedEvent.EventOrder, useOutbox: true)
+                    new UnitOfWorkEventRecord(distributedEvent.EventData.GetType(), distributedEvent.EventData, distributedEvent.EventOrder)
                 );
             }
         }

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

@@ -3,6 +3,7 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Volo.Abp.Domain;
 using Volo.Abp.EntityFrameworkCore.DependencyInjection;
+using Volo.Abp.EntityFrameworkCore.DistributedEvents;
 using Volo.Abp.Modularity;
 using Volo.Abp.Uow.EntityFrameworkCore;
 
@@ -26,6 +27,7 @@ namespace Volo.Abp.EntityFrameworkCore
             });
 
             context.Services.TryAddTransient(typeof(IDbContextProvider<>), typeof(UnitOfWorkDbContextProvider<>));
+            context.Services.AddTransient(typeof(DbContextEventOutbox<>));
         }
     }
 }

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

@@ -0,0 +1,29 @@
+using System.Threading.Tasks;
+using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.Guids;
+
+namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
+{
+    public class DbContextEventOutbox<TDbContext> : IEventOutbox 
+        where TDbContext : IHasEventOutbox
+    {
+        protected IDbContextProvider<TDbContext> DbContextProvider { get; }
+        protected IGuidGenerator GuidGenerator { get; }
+
+        public DbContextEventOutbox(
+            IDbContextProvider<TDbContext> dbContextProvider,
+            IGuidGenerator guidGenerator)
+        {
+            DbContextProvider = dbContextProvider;
+            GuidGenerator = guidGenerator;
+        }
+        
+        public async Task EnqueueAsync(string eventName, byte[] eventData)
+        {
+            var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync();
+            dbContext.OutgoingEventRecords.Add(
+                new OutgoingEventRecord(GuidGenerator.Create(), eventName, eventData)
+            );
+        }
+    }
+}

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

@@ -2,7 +2,7 @@ using Microsoft.EntityFrameworkCore;
 
 namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
 {
-    public interface IHasEventOutbox
+    public interface IHasEventOutbox : IEfCoreDbContext
     {
         DbSet<OutgoingEventRecord> OutgoingEventRecords { get; set; }
     }

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

@@ -19,9 +19,12 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
             this.SetDefaultsForExtraProperties();
         }
 
-        public OutgoingEventRecord(Guid id)
+        public OutgoingEventRecord(Guid id, string eventName, byte[] eventData)
             : base(id)
         {
+            EventName = eventName;
+            EventData = eventData;
+            
             ExtraProperties = new ExtraPropertyDictionary();
             this.SetDefaultsForExtraProperties();
         }

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

@@ -182,6 +182,11 @@ namespace Volo.Abp.EventBus.Kafka
         {
             unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
         }
+        
+        protected override byte[] Serialize(object eventData)
+        {
+            return Serializer.Serialize(eventData);
+        }
 
         public virtual async Task PublishAsync(Type eventType, object eventData, Headers headers, Dictionary<string, object> headersArguments)
         {

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

@@ -198,6 +198,11 @@ namespace Volo.Abp.EventBus.RabbitMq
         {
             unitOfWork.AddOrReplaceDistributedEvent(eventRecord);
         }
+        
+        protected override byte[] Serialize(object eventData)
+        {
+            return Serializer.Serialize(eventData);
+        }
 
         public Task PublishAsync(Type eventType, object eventData, IBasicProperties properties, Dictionary<string, object> headersArguments = null)
         {

framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/IRabbitMqSerializer.cs

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

framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs

@@ -19,6 +19,7 @@ namespace Volo.Abp.EventBus.Rebus
     public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDependency
     {
         protected IBus Rebus { get; }
+        protected IRebusSerializer Serializer { get; }
 
         //TODO: Accessing to the List<IEventHandlerFactory> may not be thread-safe!
         protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; }
@@ -32,7 +33,8 @@ namespace Volo.Abp.EventBus.Rebus
             IBus rebus,
             IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
             IOptions<AbpRebusEventBusOptions> abpEventBusRebusOptions,
-            IEventErrorHandler errorHandler) :
+            IEventErrorHandler errorHandler,
+            IRebusSerializer serializer) :
             base(
                 serviceScopeFactory,
                 currentTenant,
@@ -41,6 +43,7 @@ namespace Volo.Abp.EventBus.Rebus
                 abpDistributedEventBusOptions)
         {
             Rebus = rebus;
+            Serializer = serializer;
             AbpRebusEventBusOptions = abpEventBusRebusOptions.Value;
 
             HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
@@ -178,5 +181,10 @@ namespace Volo.Abp.EventBus.Rebus
 
             return false;
         }
+
+        protected override byte[] Serialize(object eventData)
+        {
+            return Serializer.Serialize(eventData);
+        }
     }
 }

framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/Utf8JsonRabbitMqSerializer.cs

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

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

@@ -1,4 +1,3 @@
-using System.Collections.Generic;
 using Volo.Abp.Collections;
 
 namespace Volo.Abp.EventBus.Distributed
@@ -7,16 +6,12 @@ namespace Volo.Abp.EventBus.Distributed
     {
         public ITypeList<IEventHandler> Handlers { get; }
         
-        public List<OutboxConfig> Outboxes { get; }
+        public OutboxConfigList Outboxes { get; }
 
         public AbpDistributedEventBusOptions()
         {
             Handlers = new TypeList<IEventHandler>();
+            Outboxes = new OutboxConfigList();
         }
     }
-
-    public class OutboxConfig
-    {
-        
-    }
 }

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

@@ -1,4 +1,5 @@
 using System;
+using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
@@ -31,6 +32,11 @@ namespace Volo.Abp.EventBus.Distributed
             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,
@@ -68,7 +74,29 @@ namespace Volo.Abp.EventBus.Distributed
         
         private async Task<bool> AddToOutboxAsync(Type eventType, object eventData)
         {
+            var unitOfWork = UnitOfWorkManager.Current;
+            if (unitOfWork == null)
+            {
+                return false;
+            }
+
+            foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes)
+            {
+                if (outboxConfig.Selector == null || outboxConfig.Selector(eventType))
+                {
+                    var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType);
+                    var eventName = EventNameAttribute.GetNameOrDefault(eventType);
+                    await eventOutbox.EnqueueAsync(
+                        eventName,
+                        Serialize(eventData)
+                    );
+                    return true;
+                }
+            }
+            
             return false;
         }
+
+        protected abstract byte[] Serialize(object eventData);
     }
 }

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

@@ -0,0 +1,9 @@
+using System.Threading.Tasks;
+
+namespace Volo.Abp.EventBus.Distributed
+{
+    public interface IEventOutbox
+    {
+        Task EnqueueAsync(string eventName, byte[] eventData);
+    }
+}

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

@@ -0,0 +1,18 @@
+using System;
+
+namespace Volo.Abp.EventBus.Distributed
+{
+    public class OutboxConfig
+    {
+        public string Name { get; }
+        
+        public Type ImplementationType { get; set; }
+        public Func<Type, bool> Selector { get; set; }
+
+        public OutboxConfig(string name, Type implementationType, Func<Type, bool> selector = null)
+        {
+            Name = name;
+            ImplementationType = implementationType;
+        }
+    }
+}

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

@@ -0,0 +1,8 @@
+using System.Collections.Generic;
+
+namespace Volo.Abp.EventBus.Distributed
+{
+    public class OutboxConfigList : List<OutboxConfig>
+    {
+    }
+}

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs

@@ -99,7 +99,7 @@ namespace Volo.Abp.EventBus
         }
 
         /// <inheritdoc/>
-        public async Task PublishAsync(
+        public virtual async Task PublishAsync(
             Type eventType,
             object eventData,
             bool onUnitOfWorkComplete = true)

framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs

@@ -674,8 +674,7 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
                         new UnitOfWorkEventRecord(
                             distributedEvent.EventData.GetType(),
                             distributedEvent.EventData,
-                            distributedEvent.EventOrder,
-                            useOutbox: true
+                            distributedEvent.EventOrder
                         )
                     );
                 }

framework/src/Volo.Abp.Uow/Volo/Abp/Uow/UnitOfWorkEventRecord.cs

@@ -22,7 +22,7 @@ namespace Volo.Abp.Uow
             Type eventType,
             object eventData,
             long eventOrder,
-            bool useOutbox = false)
+            bool useOutbox = true)
         {
             EventType = eventType;
             EventData = eventData;

test/DistEvents/DistDemoApp/DistDemoApp.csproj

@@ -16,6 +16,7 @@
     <ItemGroup>
         <ProjectReference Include="..\..\..\framework\src\Volo.Abp.EntityFrameworkCore.SqlServer\Volo.Abp.EntityFrameworkCore.SqlServer.csproj" />
         <ProjectReference Include="..\..\..\framework\src\Volo.Abp.Autofac\Volo.Abp.Autofac.csproj" />
+        <ProjectReference Include="..\..\..\framework\src\Volo.Abp.EventBus.RabbitMQ\Volo.Abp.EventBus.RabbitMQ.csproj" />
     </ItemGroup>
 
     <ItemGroup>

test/DistEvents/DistDemoApp/DistDemoAppModule.cs

@@ -2,14 +2,18 @@ using Microsoft.Extensions.DependencyInjection;
 using Volo.Abp.Autofac;
 using Volo.Abp.Domain.Entities.Events.Distributed;
 using Volo.Abp.EntityFrameworkCore;
+using Volo.Abp.EntityFrameworkCore.DistributedEvents;
 using Volo.Abp.EntityFrameworkCore.SqlServer;
+using Volo.Abp.EventBus.Distributed;
+using Volo.Abp.EventBus.RabbitMq;
 using Volo.Abp.Modularity;
 
 namespace DistDemoApp
 {
     [DependsOn(
         typeof(AbpEntityFrameworkCoreSqlServerModule),
-        typeof(AbpAutofacModule)
+        typeof(AbpAutofacModule),
+        typeof(AbpEventBusRabbitMqModule)
     )]
     public class DistDemoAppModule : AbpModule
     {
@@ -32,6 +36,13 @@ namespace DistDemoApp
                 options.EtoMappings.Add<TodoItem, TodoItemEto>();
                 options.AutoEventSelectors.Add<TodoItem>();
             });
+            
+            Configure<AbpDistributedEventBusOptions>(options =>
+            {
+                options.Outboxes.Add(
+                    new OutboxConfig("Default", typeof(DbContextEventOutbox<TodoDbContext>))
+                );
+            });
         }
     }
 }

test/DistEvents/DistDemoApp/Migrations/20210908063422_Added_Outbox.Designer.cs

@@ -0,0 +1,118 @@
+// <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("20210908063422_Added_Outbox")]
+    partial class Added_Outbox
+    {
+        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.OutgoingEventRecord", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    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/20210908063422_Added_Outbox.cs

@@ -0,0 +1,31 @@
+using System;
+using Microsoft.EntityFrameworkCore.Migrations;
+
+namespace DistDemoApp.Migrations
+{
+    public partial class Added_Outbox : Migration
+    {
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.CreateTable(
+                name: "AbpEventOutbox",
+                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)
+                },
+                constraints: table =>
+                {
+                    table.PrimaryKey("PK_AbpEventOutbox", x => x.Id);
+                });
+        }
+
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.DropTable(
+                name: "AbpEventOutbox");
+        }
+    }
+}

test/DistEvents/DistDemoApp/Migrations/TodoDbContextModelSnapshot.cs

@@ -87,6 +87,29 @@ namespace DistDemoApp.Migrations
 
                     b.ToTable("TodoSummaries");
                 });
+
+            modelBuilder.Entity("Volo.Abp.EntityFrameworkCore.DistributedEvents.OutgoingEventRecord", b =>
+                {
+                    b.Property<Guid>("Id")
+                        .HasColumnType("uniqueidentifier");
+
+                    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/TodoDbContext.cs

@@ -1,13 +1,15 @@
 using Microsoft.EntityFrameworkCore;
 using Volo.Abp.Domain.Entities;
 using Volo.Abp.EntityFrameworkCore;
+using Volo.Abp.EntityFrameworkCore.DistributedEvents;
 
 namespace DistDemoApp
 {
-    public class TodoDbContext : AbpDbContext<TodoDbContext>
+    public class TodoDbContext : AbpDbContext<TodoDbContext>, IHasEventOutbox
     {
         public DbSet<TodoItem> TodoItems { get; set; }
         public DbSet<TodoSummary> TodoSummaries { get; set; }
+        public DbSet<OutgoingEventRecord> OutgoingEventRecords { get; set; }
 
         public TodoDbContext(DbContextOptions<TodoDbContext> options)
             : base(options)
@@ -18,6 +20,8 @@ namespace DistDemoApp
         protected override void OnModelCreating(ModelBuilder modelBuilder)
         {
             base.OnModelCreating(modelBuilder);
+            
+            modelBuilder.ConfigureEventOutbox();
 
             modelBuilder.Entity<TodoItem>(b =>
             {

test/DistEvents/DistDemoApp/TodoEventHandler.cs

@@ -42,7 +42,7 @@ namespace DistDemoApp
             
             Console.WriteLine("Increased total count: " + todoSummary);
 
-            throw new ApplicationException("Thrown to rollback the UOW!");
+            //throw new ApplicationException("Thrown to rollback the UOW!");
         }
 
         public async Task HandleEventAsync(EntityDeletedEto<TodoItemEto> eventData)

test/DistEvents/DistDemoApp/appsettings.json

@@ -1,5 +1,16 @@
 {
     "ConnectionStrings": {
         "Default": "Server=(LocalDb)\\MSSQLLocalDB;Database=DistEventsDemo;Trusted_Connection=True"
+    },
+    "RabbitMQ": {
+        "Connections": {
+            "Default": {
+                "HostName": "localhost"
+            }
+        },
+        "EventBus": {
+            "ClientName": "DistDemoApp",
+            "ExchangeName": "DistDemo"
+        }
     }
 }