init rabbit mq

src/applications/mixcore/Program.cs

@@ -8,6 +8,7 @@
 using Mix.Lib.Middlewares;
 using Mix.Shared.Services;
 using Mix.Shared.Models.Configurations;
+using Mix.Queue.Extensions;
 var builder = MixCmsHelper.CreateWebApplicationBuilder(args);
 
 if (builder.Environment.IsDevelopment())
@@ -25,7 +26,9 @@
 var globalConfig = builder.Configuration.GetSection(MixAppSettingsSection.GlobalSettings)
                                             .Get<GlobalSettingsModel>();
 builder.Services.AddEndpointsApiExplorer();
+builder.AddMixQueue();
 builder.Services.AddMixServices(Assembly.GetExecutingAssembly(), builder.Configuration);
+
 builder.Services.ApplyMigrations(globalConfig);
 builder.Services.AddMixCors();
 builder.Services.AddScoped<MixNavigationService>();

src/applications/mixcore/wwwroot/default-mixcontent/shared/appconfigs/queue.json

@@ -11,6 +11,13 @@
     },
     "Mix": {
       "ProjectId": ""
+    },
+    "RabitMqQueueSetting": {
+      "HostName": "localhost",
+      "UserName": "",
+      "Password": "",
+      "VHost": null,
+      "Port": 5672
     }
   }
 }

src/platform/mix.constant/Enums/MixQueueProvider.cs

@@ -3,7 +3,7 @@
     public enum MixQueueProvider
     {
         GOOGLE,
-        KAFKA,
+        RABITMQ,
         AWS,
         AZURE,
         MIX

src/platform/mix.library/Publishers/MixBackgroundTaskPublisher.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Mq.Lib.Models;
 using Mix.Queue.Engines;
 
@@ -13,8 +14,9 @@ public MixBackgroundTaskPublisher(
             IMemoryQueueService<MessageQueueModel> queueService,
             IConfiguration configuration,
             MixEndpointService mixEndpointService,
-            ILogger<MixBackgroundTaskPublisher> logger)
-            : base(TopicId, queueService, configuration, mixEndpointService, logger)
+            ILogger<MixBackgroundTaskPublisher> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel> rabbitMqObjectPolicy = null)
+            : base(TopicId, queueService, configuration, mixEndpointService, logger, rabbitMqObjectPolicy)
         {
         }
     }

src/platform/mix.library/Publishers/MixDbCommandPublisher.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Mq.Lib.Models;
 using Mix.Queue.Engines;
 
@@ -13,8 +14,9 @@ public MixDbCommandPublisher(
             IMemoryQueueService<MessageQueueModel> queueService,
             IConfiguration configuration,
             MixEndpointService mixEndpointService,
-            ILogger<MixDbCommandPublisher> logger)
-            : base(TopicId, queueService, configuration, mixEndpointService, logger)
+            ILogger<MixDbCommandPublisher> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel> rabbitMqObjectPolicy = null)
+            : base(TopicId, queueService, configuration, mixEndpointService, logger, rabbitMqObjectPolicy)
         {
         }
     }

src/platform/mix.library/Publishers/MixPublisher.cs

@@ -1,10 +1,12 @@
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Mq.Lib.Models;
 using Mix.Queue.Engines;
 using Mix.Queue.Engines.MixQueue;
 using Mix.RepoDb.Publishers;
+using RabbitMQ.Client;
 
 namespace Mix.Lib.Publishers
 {
@@ -16,8 +18,9 @@ public MixPublisher(
             IMemoryQueueService<MessageQueueModel> queueService,
             IConfiguration configuration, IWebHostEnvironment environment,
             MixEndpointService mixEndpointService,
-            ILogger<MixRepoDbPublisher> logger)
-            : base(topicId, queueService, configuration, mixEndpointService, logger)
+            ILogger<MixRepoDbPublisher> logger,
+            IPooledObjectPolicy<IModel> rabbitMqObjectPolicy = null)
+            : base(topicId, queueService, configuration, mixEndpointService, logger, rabbitMqObjectPolicy)
         {
         }
     }

src/platform/mix.library/Publishers/MixViewModelChangedPublisher.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Mq.Lib.Models;
 using Mix.Queue.Engines;
 using Mix.Queue.Engines.MixQueue;
@@ -15,8 +16,9 @@ public MixViewModelChangedPublisher(
             IMemoryQueueService<MessageQueueModel> queueService,
             IConfiguration configuration,
             MixEndpointService mixEndpointService,
-            ILogger<MixViewModelChangedPublisher> logger)
-            : base(TopicId, queueService, configuration, mixEndpointService, logger)
+            ILogger<MixViewModelChangedPublisher> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel> rabbitMqObjectPolicy = null)
+            : base(TopicId, queueService, configuration, mixEndpointService, logger, rabbitMqObjectPolicy)
         {
         }
     }

src/platform/mix.library/Subscribers/MixBackgroundTaskSubscriber.cs

@@ -1,6 +1,7 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Communicator.Models;
 using Mix.Communicator.Services;
 using Mix.Database.Entities.Account;
@@ -35,8 +36,9 @@ public MixBackgroundTaskSubscriber(
             IPortalHubClientService portalHub,
             MixDbEventService mixDbEventService,
             IMemoryQueueService<MessageQueueModel> queueService,
-            ILogger<MixBackgroundTaskSubscriber> logger)
-            : base(TopicId, nameof(MixBackgroundTaskSubscriber), 20, serviceProvider, configuration, queueService, logger)
+            ILogger<MixBackgroundTaskSubscriber> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel> rabbitMqObjectPolicy = null)
+            : base(TopicId, nameof(MixBackgroundTaskSubscriber), 20, serviceProvider, configuration, queueService, logger, rabbitMqObjectPolicy)
         {
             PortalHub = portalHub;
             MixDbEventService = mixDbEventService;

src/platform/mix.library/Subscribers/MixDbCommandSubscriber.cs

@@ -1,6 +1,7 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Communicator.Models;
 using Mix.Communicator.Services;
 using Mix.Database.Entities.MixDb;
@@ -26,8 +27,9 @@ public MixDbCommandSubscriber(
             IServiceProvider serviceProvider,
             IConfiguration configuration,
             IMemoryQueueService<MessageQueueModel> queueService,
-            ILogger<MixDbCommandSubscriber> logger)
-            : base(TopicId, nameof(MixDbCommandSubscriber), 20, serviceProvider, configuration, queueService, logger)
+            ILogger<MixDbCommandSubscriber> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel> rabbitMqObjectPolicy = null)
+            : base(TopicId, nameof(MixDbCommandSubscriber), 20, serviceProvider, configuration, queueService, logger, rabbitMqObjectPolicy)
         {
         }
 

src/platform/mix.library/Subscribers/MixViewModelChangedSubscriber.cs

@@ -1,6 +1,7 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Heart.Model;
 using Mix.Lib.Interfaces;
 using Mix.Lib.Subscribers.Handlers.MixViewModelChangedHandlers;
@@ -20,8 +21,9 @@ public MixViewModelChangedSubscriber(
             IConfiguration configuration,
             IMixTenantService mixTenantService,
             IMemoryQueueService<MessageQueueModel> queueService,
-            ILogger<MixViewModelChangedSubscriber> logger)
-            : base(TopicId, nameof(MixDbCommandSubscriber), 20, serviceProvider, configuration, queueService, logger)
+            ILogger<MixViewModelChangedSubscriber> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel> rabbitMqObjectPolicy = null)
+            : base(TopicId, nameof(MixDbCommandSubscriber), 20, serviceProvider, configuration, queueService, logger, rabbitMqObjectPolicy)
         {
             _mixTenantService = mixTenantService;
         }

src/platform/mix.log/Publishers/MixLogPublisher.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Constant.Constants;
 using Mix.Mq.Lib.Models;
 using Mix.Queue.Engines;
@@ -17,8 +18,9 @@ public MixLogPublisher(
             IMemoryQueueService<MessageQueueModel> queueService,
             IConfiguration configuration,
             MixEndpointService mixEndpointService,
-            ILogger<MixLogPublisher> logger)
-            : base(TopicId, queueService, configuration, mixEndpointService, logger)
+            ILogger<MixLogPublisher> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel>? rabbitMqObjectPolicy = null)
+            : base(TopicId, queueService, configuration, mixEndpointService, logger, rabbitMqObjectPolicy)
         {
         }
     }

src/platform/mix.log/Subscribers/MixLogSubscriber.cs

@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Constant.Constants;
 using Mix.Constant.Enums;
 using Mix.Database.Entities.Queue;
@@ -42,8 +43,9 @@ public MixLogSubscriber(
             IMemoryQueueService<MessageQueueModel> queueService,
             IMixQueueLog queueMessageLogService,
             IAuditLogService auditLogService,
-            ILogger<MixLogSubscriber> logger)
-            : base(TopicId, nameof(MixLogSubscriber), 20, serviceProvider, configuration, queueService, logger)
+            ILogger<MixLogSubscriber> logger,
+            IPooledObjectPolicy<RabbitMQ.Client.IModel> rabbitMqObjectPolicy = null)
+            : base(TopicId, nameof(MixLogSubscriber), 20, serviceProvider, configuration, queueService, logger, rabbitMqObjectPolicy)
         {
             _queueMessageLogService = queueMessageLogService;
             _portalHub = portalHub;

src/platform/mix.queue/Engines/Azure/AzureQueuePublisher.cs

@@ -19,7 +19,7 @@ internal class AzureQueuePublisher<T> : IQueuePublisher<T>
         private static ServiceBusClient _client;
         private readonly AzureQueueSetting _queueSetting;
 
-        public AzureQueuePublisher(QueueSetting queueSetting, string topicName)
+        public AzureQueuePublisher(IQueueSetting queueSetting, string topicName)
         {
             _queueSetting = queueSetting as AzureQueueSetting;
             InitializeQueue(topicName);

src/platform/mix.queue/Engines/Azure/AzureQueueSubscriber.cs

@@ -19,7 +19,7 @@ internal class AzureQueueSubscriber<T> : IQueueSubscriber
         private readonly AzureQueueSetting _queueSetting;
         private readonly Func<T, Task> _messageHandler;
         public AzureQueueSubscriber(
-            QueueSetting queueSetting,
+            IQueueSetting queueSetting,
             string topicId,
             string subscriptionId,
             Func<T, Task> messageHandler)

src/platform/mix.queue/Engines/Google/GoogleQueuePublisher.cs

@@ -18,7 +18,7 @@ internal class GoogleQueuePublisher<T> : IQueuePublisher<T>
         private PublisherClient _publisher;
         private readonly GoogleQueueSetting _queueSetting;
 
-        public GoogleQueuePublisher(QueueSetting queueSetting, string topicName)
+        public GoogleQueuePublisher(IQueueSetting queueSetting, string topicName)
         {
             _queueSetting = queueSetting as GoogleQueueSetting;
             InitializeQueue(topicName);

src/platform/mix.queue/Engines/Google/GoogleQueueSubscriber.cs

@@ -23,7 +23,7 @@ internal class GoogleQueueSubscriber<T> : IQueueSubscriber
         private readonly Func<T, Task> _messageHandler;
 
         public GoogleQueueSubscriber(
-            QueueSetting queueSetting,
+            IQueueSetting queueSetting,
             string topicId,
             string subscriptionId,
             Func<T, Task> messageHandler)

src/platform/mix.queue/Engines/Mix/MixQueuePublisher.cs

@@ -19,7 +19,7 @@ public class MixQueuePublisher<T> : IQueuePublisher<T>
         private readonly MixEndpointService _mixEndpointService;
         private GrpcChannelModel<MixMq.MixMqClient> _mixMqPublisher;
 
-        public MixQueuePublisher(QueueSetting queueSetting, string topicName, MixEndpointService mixEndpointService)
+        public MixQueuePublisher(IQueueSetting queueSetting, string topicName, MixEndpointService mixEndpointService)
         {
             _topicId = topicName;
             _mixEndpointService = mixEndpointService;

src/platform/mix.queue/Engines/Mix/MixQueueSubscriber.cs

@@ -20,7 +20,7 @@
         public string SubscriptionId { get; set; }
         public bool IsProcessing { get; private set; }
         private readonly string _subscriptionId;
         private MixTopicModel<T> _topic;
         private readonly MixQueueSetting _queueSetting;
         private readonly Func<T, Task> _messageHandler;
         private readonly IMemoryQueueService<MessageQueueModel> _memQueues;
@@ -28,9 +28,9 @@
         private GrpcChannelModel<MixMq.MixMqClient> _mixMqSubscriber;
         private SubscribeRequest _subscribeRequest;
         private AsyncServerStreamingCall<SubscribeReply> _call;
         private CancellationToken _startCancellationToken;
         public MixQueueSubscriber(
-            QueueSetting queueSetting,
+            IQueueSetting queueSetting,
             string topicId,
             string subscriptionId,
             Func<T, Task> messageHandler,

src/platform/mix.queue/Engines/PublisherBase.cs

@@ -1,12 +1,14 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.ObjectPool;
 using Mix.Heart.Exceptions;
 using Mix.Mq.Lib.Models;
 using Mix.Queue.Engines.MixQueue;
 using Mix.Queue.Interfaces;
 using Mix.Queue.Models.QueueSetting;
 using Mix.Shared.Services;
+using RabbitMQ.Client;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -25,22 +27,25 @@ public abstract class PublisherBase : BackgroundService
         private readonly string _topicId;
         private MixQueueProvider _provider;
         protected ILogger<PublisherBase> _logger;
+        private readonly IPooledObjectPolicy<IModel> _rabbitMqObjectPolicy;
         protected PublisherBase(
             string topicId,
             IMemoryQueueService<MessageQueueModel> queueService,
             IConfiguration configuration,
             MixEndpointService mixEndpointService,
-            ILogger<PublisherBase> logger)
+            ILogger<PublisherBase> logger,
+            IPooledObjectPolicy<IModel> rabbitMqObjectPolicy)
         {
             _queueService = queueService;
             _configuration = configuration;
             _topicId = topicId;
             _mixEndpointService = mixEndpointService;
             _logger = logger;
+            _rabbitMqObjectPolicy = rabbitMqObjectPolicy;
         }
 
         private List<IQueuePublisher<MessageQueueModel>> CreatePublisher(
-            string topicName)
+            string topicId)
         {
             try
             {
@@ -62,7 +67,7 @@ private List<IQueuePublisher<MessageQueueModel>> CreatePublisher(
 
                         queuePublishers.Add(
                             QueueEngineFactory.CreatePublisher<MessageQueueModel>(
-                                _provider, azureSetting, topicName, _mixEndpointService));
+                                _provider, azureSetting, topicId, _mixEndpointService));
                         break;
                     case MixQueueProvider.GOOGLE:
                         var googleSettingPath = _configuration.GetSection("MessageQueueSetting:GoogleQueueSetting");
@@ -72,16 +77,22 @@ private List<IQueuePublisher<MessageQueueModel>> CreatePublisher(
 
                         queuePublishers.Add(
                             QueueEngineFactory.CreatePublisher<MessageQueueModel>(
-                                _provider, googleSetting, topicName, _mixEndpointService));
+                                _provider, googleSetting, topicId, _mixEndpointService));
                         break;
+
+                    case MixQueueProvider.RABITMQ:
+                        queuePublishers.Add(
+                            QueueEngineFactory.CreateRabbitMqPublisher<MessageQueueModel>(_rabbitMqObjectPolicy, topicId));
+                        break;
+
                     case MixQueueProvider.MIX:
                         if (_mixEndpointService.MixMq != null)
                         {
                             var mixSettingPath = _configuration.GetSection("MessageQueueSetting:Mix");
                             var mixSetting = new MixQueueSetting();
                             mixSettingPath.Bind(mixSetting);
                             queuePublishers.Add(
-                               QueueEngineFactory.CreatePublisher<MessageQueueModel>(_provider, mixSetting, topicName, _mixEndpointService));
+                               QueueEngineFactory.CreatePublisher<MessageQueueModel>(_provider, mixSetting, topicId, _mixEndpointService));
                         }
                         break;
                 }