Created RabbitMq connection and channel pooling draft.

framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/RabbitMqBackgroundJobManager.cs

@@ -0,0 +1,51 @@
+using System;
+using System.Threading.Tasks;
+using RabbitMQ.Client;
+using Volo.Abp.DependencyInjection;
+using Volo.Abp.RabbitMQ;
+
+namespace Volo.Abp.BackgroundJobs.RabbitMQ
+{
+    public class RabbitMqBackgroundJobManager : IBackgroundJobManager, ITransientDependency
+    {
+        protected IChannelPool ChannelPool { get; }
+        protected IRabbitMqSerializer Serializer { get; }
+
+        public RabbitMqBackgroundJobManager(IChannelPool channelPool, IRabbitMqSerializer serializer)
+        {
+            Serializer = serializer;
+            ChannelPool = channelPool;
+        }
+
+        public Task<string> EnqueueAsync<TArgs>(TArgs args, BackgroundJobPriority priority = BackgroundJobPriority.Normal,
+            TimeSpan? delay = null)
+        {
+            var jobName = BackgroundJobNameAttribute.GetName<TArgs>();
+            var queueName = "BackgroundJobs." + jobName; //TODO: Make prefix optional
+
+            using (var channelAccessor = ChannelPool.Acquire(queueName))
+            {
+                var properties = channelAccessor.Channel.CreateBasicProperties();
+                properties.Persistent = true;
+
+                Publish(channelAccessor.Channel, queueName, args, properties);
+            }
+
+            return null; //TODO: Can RabbitMQ return a message identifier?
+        }
+
+        private void Publish<TArgs>(
+            IModel channel, 
+            string queueName, 
+            TArgs args, 
+            IBasicProperties properties)
+        {
+            channel.BasicPublish(
+                exchange: "",
+                routingKey: queueName,
+                basicProperties: properties,
+                body: Serializer.Serialize(args)
+            );
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo.Abp.RabbitMQ.csproj

@@ -15,7 +15,7 @@
 
   <ItemGroup>
     <PackageReference Include="RabbitMQ.Client" Version="5.1.0" />
-    <ProjectReference Include="..\Volo.Abp.Core\Volo.Abp.Core.csproj" />
+    <ProjectReference Include="..\Volo.Abp.Json\Volo.Abp.Json.csproj" />
   </ItemGroup>
 
 </Project>

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpBackgroundJobsRabbitMqModule.cs

@@ -1,8 +1,12 @@
 using Microsoft.Extensions.DependencyInjection;
+using Volo.Abp.Json;
 using Volo.Abp.Modularity;
 
 namespace Volo.Abp.RabbitMQ
 {
+    [DependsOn(
+        typeof(AbpJsonModule)
+        )]
     public class AbpRabbitMqModule : AbpModule
     {
         public override void ConfigureServices(ServiceConfigurationContext context)

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqOptions.cs

@@ -0,0 +1,12 @@
+namespace Volo.Abp.RabbitMQ
+{
+    public class AbpRabbitMqOptions
+    {
+        public RabbitMqConnections Connections { get; }
+
+        public AbpRabbitMqOptions()
+        {
+            Connections = new RabbitMqConnections();
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs

@@ -0,0 +1,44 @@
+using RabbitMQ.Client;
+using Volo.Abp.DependencyInjection;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public class ChannelPool : IChannelPool, ISingletonDependency
+    {
+        protected IConnectionPool ConnectionPool { get; }
+
+        public ChannelPool(IConnectionPool connectionPool)
+        {
+            ConnectionPool = connectionPool;
+        }
+
+        public virtual IChannelAccessor Acquire(string channelName = null)
+        {
+            //TODO: Pool channels!
+            return new ChannelAccessor(
+                CreateChannel(channelName)
+            );
+        }
+
+        protected virtual IModel CreateChannel(string channelName)
+        {
+            //TODO: How to determine the right connection name?
+            return ConnectionPool.Get().CreateModel();
+        }
+
+        protected class ChannelAccessor : IChannelAccessor
+        {
+            public IModel Channel { get; }
+
+            public ChannelAccessor(IModel channel)
+            {
+                Channel = channel;
+            }
+
+            public void Dispose()
+            {
+                Channel.Dispose();
+            }
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ConnectionPool.cs

@@ -0,0 +1,32 @@
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using Microsoft.Extensions.Options;
+using RabbitMQ.Client;
+using Volo.Abp.DependencyInjection;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public class ConnectionPool : IConnectionPool, ISingletonDependency
+    {
+        protected AbpRabbitMqOptions Options { get; }
+
+        protected ConcurrentDictionary<string, IConnection> Connections { get; }
+
+        public ConnectionPool(IOptions<AbpRabbitMqOptions> options)
+        {
+            Options = options.Value;
+            Connections = new ConcurrentDictionary<string, IConnection>();
+        }
+
+        public virtual IConnection Get(string connectionName = null)
+        {
+            return Connections.GetOrAdd(connectionName, () =>
+            {
+                var connectionFactory = Options.Connections.GetOrDefault(connectionName)
+                                        ?? Options.Connections.Default;
+
+                return connectionFactory.CreateConnection();
+            });
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IChannelAccessor.cs

@@ -0,0 +1,10 @@
+using System;
+using RabbitMQ.Client;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public interface IChannelAccessor : IDisposable
+    {
+        IModel Channel { get; }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IChannelPool.cs

@@ -0,0 +1,7 @@
+namespace Volo.Abp.RabbitMQ
+{
+    public interface IChannelPool
+    {
+        IChannelAccessor Acquire(string channelName = null);
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IConnectionPool.cs

@@ -0,0 +1,9 @@
+using RabbitMQ.Client;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public interface IConnectionPool
+    {
+        IConnection Get(string connectionName = null);
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IRabbitMqSerializer.cs

@@ -0,0 +1,11 @@
+using System;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public interface IRabbitMqSerializer
+    {
+        byte[] Serialize(object obj);
+
+        object Deserialize(byte[] value, Type type);
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/RabbitMqConnections.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Collections.Generic;
+using JetBrains.Annotations;
+using RabbitMQ.Client;
+
+namespace Volo.Abp.RabbitMQ
+{
+    [Serializable]
+    public class RabbitMqConnections : Dictionary<string, ConnectionFactory>
+    {
+        public const string DefaultConnectionName = "Default";
+        
+        [NotNull]
+        public ConnectionFactory Default
+        {
+            get => this.GetOrDefault(DefaultConnectionName);
+            set => this[DefaultConnectionName] = Check.NotNull(value, nameof(value));
+        }
+
+        public RabbitMqConnections()
+        {
+            Default = new ConnectionFactory();
+        }
+    }
+}

framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/Utf8JsonRabbitMqSerializer.cs

@@ -0,0 +1,32 @@
+using System;
+using System.Text;
+using Volo.Abp.DependencyInjection;
+using Volo.Abp.Json;
+
+namespace Volo.Abp.RabbitMQ
+{
+    public class Utf8JsonRabbitMqSerializer : IRabbitMqSerializer, ITransientDependency
+    {
+        private readonly IJsonSerializer _jsonSerializer;
+
+        public Utf8JsonRabbitMqSerializer(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>(string value)
+        {
+            return _jsonSerializer.Deserialize<T>(value);
+        }
+    }
+}