save to outbox

katmacaoglu · Sep 8, 2021 · d1ea473 · d1ea473
1 parent 5f50d33
commit d1ea473
Show file tree

Hide file tree

Showing 26 changed files with 371 additions and 18 deletions.
diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/AbpDbContext.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/AbpDbContext.cs
@@ -201,7 +201,7 @@ private void PublishEntityEvents(EntityEventReport changeReport)
             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)
                 );
             }
         }

diff --git a/...Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/AbpEntityFrameworkCoreModule.cs b/...Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/AbpEntityFrameworkCoreModule.cs
@@ -3,6 +3,7 @@
 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 @@ public override void ConfigureServices(ServiceConfigurationContext context)
             });
 
             context.Services.TryAddTransient(typeof(IDbContextProvider<>), typeof(UnitOfWorkDbContextProvider<>));
+            context.Services.AddTransient(typeof(DbContextEventOutbox<>));
         }
     }
 }
diff --git a/...ntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventOutbox.cs b/...ntityFrameworkCore/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)
+            );
+        }
+    }
+}
diff --git a/...Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IHasEventOutbox.cs b/...Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IHasEventOutbox.cs
@@ -2,7 +2,7 @@
 
 namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
 {
-    public interface IHasEventOutbox
+    public interface IHasEventOutbox : IEfCoreDbContext
     {
         DbSet<OutgoingEventRecord> OutgoingEventRecords { get; set; }
     }

diff --git a/...EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/OutgoingEventRecord.cs b/...EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/OutgoingEventRecord.cs
@@ -19,9 +19,12 @@ protected OutgoingEventRecord()
             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();
         }

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
@@ -182,6 +182,11 @@ protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventR
         {
             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)
         {

diff --git a/.../src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/.../src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs
@@ -198,6 +198,11 @@ protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventR
         {
             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)
         {

diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/IRabbitMqSerializer.cs b/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);
+    }
+}
diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs b/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 @@ public RebusDistributedEventBus(
             IBus rebus,
             IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions,
             IOptions<AbpRebusEventBusOptions> abpEventBusRebusOptions,
-            IEventErrorHandler errorHandler) :
+            IEventErrorHandler errorHandler,
+            IRebusSerializer serializer) :
             base(
                 serviceScopeFactory,
                 currentTenant,
@@ -41,6 +43,7 @@ public RebusDistributedEventBus(
                 abpDistributedEventBusOptions)
         {
             Rebus = rebus;
+            Serializer = serializer;
             AbpRebusEventBusOptions = abpEventBusRebusOptions.Value;
 
             HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
@@ -178,5 +181,10 @@ private static bool ShouldTriggerEventForHandler(Type targetEventType, Type hand
 
             return false;
         }
+
+        protected override byte[] Serialize(object eventData)
+        {
+            return Serializer.Serialize(eventData);
+        }
     }
 }
diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/Utf8JsonRabbitMqSerializer.cs b/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));
+        }
+    }
+}
diff --git a/...work/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/AbpDistributedEventBusOptions.cs b/...work/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 @@ public class AbpDistributedEventBusOptions
     {
         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
-    {
-
-    }
 }
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
@@ -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 @@ public IDisposable Subscribe<TEvent>(IDistributedEventHandler<TEvent> handler) w
             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 @@ public async Task PublishAsync(
 
         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);
     }
 }
diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IEventOutbox.cs b/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);
+    }
+}
diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxConfig.cs b/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;
+        }
+    }
+}
diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/OutboxConfigList.cs b/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>
+    {
+    }
+}
diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs
@@ -99,7 +99,7 @@ public Task PublishAsync<TEvent>(TEvent eventData, bool onUnitOfWorkComplete = t
         }
 
         /// <inheritdoc/>
-        public async Task PublishAsync(
+        public virtual async Task PublishAsync(
             Type eventType,
             object eventData,
             bool onUnitOfWorkComplete = true)

diff --git a/framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs b/framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs
@@ -674,8 +674,7 @@ protected virtual void TriggerDomainEvents(object entity)
                         new UnitOfWorkEventRecord(
                             distributedEvent.EventData.GetType(),
                             distributedEvent.EventData,
-                            distributedEvent.EventOrder,
-                            useOutbox: true
+                            distributedEvent.EventOrder
                         )
                     );
                 }

diff --git a/framework/src/Volo.Abp.Uow/Volo/Abp/Uow/UnitOfWorkEventRecord.cs b/framework/src/Volo.Abp.Uow/Volo/Abp/Uow/UnitOfWorkEventRecord.cs
@@ -22,7 +22,7 @@ public UnitOfWorkEventRecord(
             Type eventType,
             object eventData,
             long eventOrder,
-            bool useOutbox = false)
+            bool useOutbox = true)
         {
             EventType = eventType;
             EventData = eventData;

diff --git a/test/DistEvents/DistDemoApp/DistDemoApp.csproj b/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>

diff --git a/test/DistEvents/DistDemoApp/DistDemoAppModule.cs b/test/DistEvents/DistDemoApp/DistDemoAppModule.cs
@@ -2,14 +2,18 @@
 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 @@ public override void ConfigureServices(ServiceConfigurationContext context)
                 options.EtoMappings.Add<TodoItem, TodoItemEto>();
                 options.AutoEventSelectors.Add<TodoItem>();
             });
+
+            Configure<AbpDistributedEventBusOptions>(options =>
+            {
+                options.Outboxes.Add(
+                    new OutboxConfig("Default", typeof(DbContextEventOutbox<TodoDbContext>))
+                );
+            });
         }
     }
 }