增加MQ

ds-wms-service/DS.Module.Core/DS.Module.Core.csproj

@@ -22,6 +22,7 @@
       <PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="8.0.4" />
       <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="8.0.0" />
       <PackageReference Include="NLog" Version="5.2.8" />
+      <PackageReference Include="SharpZipLib" Version="1.4.2" />
       <PackageReference Include="Snowflake.Core" Version="2.0.0" />
       <PackageReference Include="SqlSugarCore" Version="5.1.4.136" />
       <PackageReference Include="Swashbuckle.AspNetCore" Version="6.5.0" />

ds-wms-service/DS.Module.Core/Extensions/SharpZipLib.cs

@@ -0,0 +1,56 @@
+using ICSharpCode.SharpZipLib.BZip2;
+using Org.BouncyCastle.Utilities.Bzip2;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace DS.Module.Core.Extensions
+{
+    public class SharpZipLib
+    {
+
+        /// <summary>
+        /// 数据压缩
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        public static string Compress(string input)
+        {
+            string result = string.Empty;
+            byte[] buffer = Encoding.UTF8.GetBytes(input);
+            using (MemoryStream outputStream = new MemoryStream())
+            {
+                using (BZip2OutputStream zipStream = new BZip2OutputStream(outputStream))
+                {
+                    zipStream.Write(buffer, 0, buffer.Length);
+                    zipStream.Close();
+                }
+                return Convert.ToBase64String(outputStream.ToArray());
+            }
+        }
+        /// <summary>
+        /// 数据解压缩
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        public static string Decompress(string input)
+        {
+            string result = string.Empty;
+            byte[] buffer = Convert.FromBase64String(input);
+            using (Stream inputStream = new MemoryStream(buffer))
+            {
+                BZip2InputStream zipStream = new BZip2InputStream(inputStream);
+
+                using (StreamReader reader = new StreamReader(zipStream, Encoding.UTF8))
+                {
+                    //输出
+                    result = reader.ReadToEnd();
+                }
+            }
+
+            return result;
+        }
+    }
+}

ds-wms-service/DS.Module.MQ/IRabbitMQService.cs

@@ -1,12 +1,24 @@
-using System;
+using RabbitMQ.Client;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 
+using DS.Module.Core;
+
 namespace DS.Module.MQ
 {
+    /// <summary>
+    /// 
+    /// </summary>
     public interface IRabbitMQService
     {
+        /// <summary>
+        /// 发送MQ报文
+        /// </summary>
+        /// <param name="request">请求参数</param>
+        /// <returns>true-发送成功 false-发送失败</returns>
+        Task<DataResult> PublishMessage(MQPublishMessageReqDto request);
     }
 }

ds-wms-service/DS.Module.MQ/MQPublishMessageReqDto.cs

@@ -0,0 +1,44 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace DS.Module.MQ
+{
+    /// <summary>
+    /// 
+    /// </summary>
+    public class MQPublishMessageReqDto
+    {
+        /// <summary>
+        /// MQ URI
+        /// </summary>
+        public string mqUri { get; set; }
+
+        /// <summary>
+        /// Exchange Name
+        /// </summary>
+        public string mqExchangeName { get; set; }
+
+        /// <summary>
+        /// Queue Name
+        /// </summary>
+        public string mqQueueName { get; set; }
+
+        /// <summary>
+        /// 业务主键
+        /// </summary>
+        public string BusinessId { get; set; }
+
+        /// <summary>
+        /// 请求MQ报文（JSON串行化）
+        /// </summary>
+        public string json { get; set; }
+
+        /// <summary>
+        /// 是否默认压缩
+        /// </summary>
+        public bool IsZip { get; set; } = false;
+    }
+}

ds-wms-service/DS.Module.MQ/RabbitMQService.cs

@@ -1,12 +1,87 @@
-using System;
+using NLog;
+using RabbitMQ.Client;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
+using DS.Module.Core.Extensions;
+using static AnyDiff.DifferenceLines;
+using DS.Module.Core;
+using Microsoft.VisualBasic;
 
 namespace DS.Module.MQ
 {
     public class RabbitMQService: IRabbitMQService
     {
+        private static readonly Logger _logger = LogManager.GetCurrentClassLogger();
+
+        /// <summary>
+        /// 发送MQ报文
+        /// </summary>
+        /// <param name="request">请求参数</param>
+        /// <returns>true-发送成功 false-发送失败</returns>
+        public async Task<DataResult> PublishMessage(MQPublishMessageReqDto request)
+        {
+            bool isException = false;
+            string msg = string.Empty;
+
+            try
+            {
+                ConnectionFactory factory = new ConnectionFactory();
+                factory.Uri = new Uri(request.mqUri);
+
+                using (IConnection conn = factory.CreateConnection())
+                {
+
+                    try
+                    {
+                        IModel mqModel = conn.CreateModel();
+                        mqModel.ExchangeDeclare(request.mqExchangeName, ExchangeType.Direct);
+
+                        //var queueName = $"{MqActionQueueName}.{(item.SubTenantId.HasValue && item.SubTenantId > 0 ? item.SubTenantId.Value : item.TenantId)}";
+                        mqModel.QueueDeclare(request.mqQueueName, false, false, false, null);
+                        mqModel.QueueBind(request.mqQueueName, request.mqExchangeName, request.mqQueueName, null);
+
+
+                        byte[] messageBodyBytes = Encoding.UTF8.GetBytes(request.IsZip ? SharpZipLib.Compress(request.json) : request.json);
+
+
+                        IBasicProperties props = mqModel.CreateBasicProperties();
+                        props.DeliveryMode = 2;
+                        mqModel.BasicPublish(request.mqExchangeName, request.mqQueueName, props, messageBodyBytes);
+
+
+                        _logger.Info($"Id:{request.BusinessId}，订舱数据回推，已发送数据到消息队列【{request.mqUri}】，队列名称：【{request.mqQueueName}】");
+
+                    }
+                    catch (Exception inne)
+                    {
+                        _logger.Error($"Id:{request.BusinessId}，推送MQ时发生异常，原因:{inne.Message}");
+
+                        msg = $"Id:{request.BusinessId}，推送MQ时发生异常，原因:{inne.Message}";
+
+                        isException = true;
+                    }
+                    finally {
+                        conn.Close();
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.Error($"Id:{request.BusinessId},启动推送MQ时发生异常，原因:{ex.Message}");
+
+                msg = $"Id:{request.BusinessId},启动推送MQ时发生异常，原因:{ex.Message}";
+
+                isException = true;
+            }
+
+            if(isException)
+                return DataResult.Failed(msg);
+
+            return DataResult.Successed("发送成功");
+        }
+
     }
 }

ds-wms-service/DS.Module.Nuget/DS.Module.Nuget.csproj

@@ -47,6 +47,7 @@
     <PackageReference Include="Npoi.Mapper" Version="6.2.1" />
     <PackageReference Include="RabbitMQ.Client" Version="6.5.0" />
     <PackageReference Include="RazorEngineCore" Version="2024.4.1" />
+    <PackageReference Include="SharpZipLib" Version="1.4.2" />
     <PackageReference Include="Snowflake.Core" Version="2.0.0" />
     <PackageReference Include="Snowflake.Data" Version="2.2.0" />
     <PackageReference Include="SqlSugar.IOC" Version="2.0.0" />