Improve performance

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

@@ -57,35 +57,31 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
         }
 
         [UnitOfWork]
-        public async Task MarkAsProcessedAsync(Guid id)
+        public virtual async Task MarkAsProcessedAsync(Guid id)
         {
-            //TODO: Optimize?
             var dbContext = await DbContextProvider.GetDbContextAsync();
-            var incomingEvent = await dbContext.IncomingEvents.FindAsync(id);
+            var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName();
-            if (incomingEvent != null)
+
-            {
+            var sql = $"UPDATE {tableName} SET Processed = 1, ProcessedTime = '{Clock.Now}' WHERE Id = '{id}'";
-                incomingEvent.MarkAsProcessed(Clock.Now);
+            await dbContext.Database.ExecuteSqlRawAsync(sql);
-            }
         }
 
         [UnitOfWork]
-        public async Task<bool> ExistsByMessageIdAsync(string messageId)
+        public virtual async Task<bool> ExistsByMessageIdAsync(string messageId)
         {
-            //TODO: Optimize
             var dbContext = await DbContextProvider.GetDbContextAsync();
             return await dbContext.IncomingEvents.AnyAsync(x => x.MessageId == messageId);
         }
 
         [UnitOfWork]
-        public async Task DeleteOldEventsAsync()
+        public virtual async Task DeleteOldEventsAsync()
         {
-            //TODO: Optimize
             var dbContext = await DbContextProvider.GetDbContextAsync();
+            var tableName = dbContext.IncomingEvents.EntityType.GetSchemaQualifiedTableName();
             var timeToKeepEvents = Clock.Now.Add(DistributedEventsOptions.InboxKeepEventTimeSpan);
-            var oldEvents = await dbContext.IncomingEvents
+
-                .Where(x => x.Processed && x.CreationTime < timeToKeepEvents)
+            var sql = $"DELETE FROM {tableName} WHERE Processed = 1 AND CreationTime < '{timeToKeepEvents}'";
-                .ToListAsync();
+            await dbContext.Database.ExecuteSqlRawAsync(sql);
-            dbContext.IncomingEvents.RemoveRange(oldEvents);
         }
     }
 }

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

@@ -49,13 +49,11 @@ namespace Volo.Abp.EntityFrameworkCore.DistributedEvents
         [UnitOfWork]
         public virtual async Task DeleteAsync(Guid id)
         {
-            //TODO: Optimize?
             var dbContext = (IHasEventOutbox) await DbContextProvider.GetDbContextAsync();
-            var outgoingEvent = await dbContext.OutgoingEvents.FindAsync(id);
+            var tableName = dbContext.OutgoingEvents.EntityType.GetSchemaQualifiedTableName();
-            if (outgoingEvent != null)
+
-            {
+            var sql = $"DELETE FROM {tableName} WHERE Id = '{id}'";
-                dbContext.Remove(outgoingEvent);
+            await dbContext.Database.ExecuteSqlRawAsync(sql);
-            }
         }
     }
 }

framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs

@@ -68,34 +68,32 @@ namespace Volo.Abp.MongoDB.DistributedEvents
         }
 
         [UnitOfWork]
-        public async Task MarkAsProcessedAsync(Guid id)
+        public virtual async Task MarkAsProcessedAsync(Guid id)
         {
             var dbContext = await DbContextProvider.GetDbContextAsync();
-            var incomingEvent = await dbContext.IncomingEvents.Find(x => x.Id.Equals(id)).FirstOrDefaultAsync();
-            if (incomingEvent != null)
-            {
-                incomingEvent.MarkAsProcessed(Clock.Now);
 
-                if (dbContext.SessionHandle != null)
+            var filter = Builders<IncomingEventRecord>.Filter.Eq(x => x.Id, id);
-                {
+            var update = Builders<IncomingEventRecord>.Update.Set(x => x.Processed, true).Set(x => x.ProcessedTime, Clock.Now);
-                    await dbContext.IncomingEvents.ReplaceOneAsync(dbContext.SessionHandle, Builders<IncomingEventRecord>.Filter.Eq(e => e.Id, incomingEvent.Id), incomingEvent);
+
-                }
+            if (dbContext.SessionHandle != null)
-                else
+            {
-                {
+                await dbContext.IncomingEvents.UpdateOneAsync(dbContext.SessionHandle, filter, update);
-                    await dbContext.IncomingEvents.ReplaceOneAsync(Builders<IncomingEventRecord>.Filter.Eq(e => e.Id, incomingEvent.Id), incomingEvent);
+            }
-                }
+            else
+            {
+                await dbContext.IncomingEvents.UpdateOneAsync(filter, update);
             }
         }
 
         [UnitOfWork]
-        public async Task<bool> ExistsByMessageIdAsync(string messageId)
+        public virtual async Task<bool> ExistsByMessageIdAsync(string messageId)
         {
             var dbContext = await DbContextProvider.GetDbContextAsync();
             return await dbContext.IncomingEvents.AsQueryable().AnyAsync(x => x.MessageId == messageId);
         }
 
         [UnitOfWork]
-        public async Task DeleteOldEventsAsync()
+        public virtual async Task DeleteOldEventsAsync()
         {
             var dbContext = await DbContextProvider.GetDbContextAsync();
             var timeToKeepEvents = Clock.Now.Add(DistributedEventsOptions.InboxKeepEventTimeSpan);

framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventOutbox.cs

@@ -21,7 +21,7 @@ namespace Volo.Abp.MongoDB.DistributedEvents
         }
 
         [UnitOfWork]
-        public async Task EnqueueAsync(OutgoingEventInfo outgoingEvent)
+        public virtual async Task EnqueueAsync(OutgoingEventInfo outgoingEvent)
         {
             var dbContext = (IHasEventOutbox) await MongoDbContextProvider.GetDbContextAsync();
             if (dbContext.SessionHandle != null)
@@ -40,7 +40,7 @@ namespace Volo.Abp.MongoDB.DistributedEvents
         }
 
         [UnitOfWork]
-        public async Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default)
+        public virtual async Task<List<OutgoingEventInfo>> GetWaitingEventsAsync(int maxCount, CancellationToken cancellationToken = default)
         {
             var dbContext = (IHasEventOutbox) await MongoDbContextProvider.GetDbContextAsync(cancellationToken);
 
@@ -56,7 +56,7 @@ namespace Volo.Abp.MongoDB.DistributedEvents
         }
 
         [UnitOfWork]
-        public async Task DeleteAsync(Guid id)
+        public virtual async Task DeleteAsync(Guid id)
         {
             var dbContext = (IHasEventOutbox) await MongoDbContextProvider.GetDbContextAsync();
             if (dbContext.SessionHandle != null)