From 4cbee59acb483983809ee3e1243538fffc43ce99 Mon Sep 17 00:00:00 2001
From: huangxianguo <huangxianguo@haijumasstransfer.com>
Date: Wed, 19 Jun 2024 19:28:48 +0800
Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0Minio=E5=B7=A5=E5=85=B7?=
 =?UTF-8?q?=E7=B1=BB?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 MasstransferCommon/Utils/JsonUtil.cs          | 12 +++
 .../Stat/HeartbeatExporter.cs                 | 22 +++++
 .../MasstransferInfrastructure.csproj         |  1 +
 .../Minio/MinioHelper.cs                      | 88 +++++++++++++++++++
 .../Mqtt/Client/MessageQueueHelper.cs         | 16 ++--
 5 files changed, 133 insertions(+), 6 deletions(-)
 create mode 100644 MasstransferExporter/Stat/HeartbeatExporter.cs
 create mode 100644 MasstransferInfrastructure/Minio/MinioHelper.cs

diff --git a/MasstransferCommon/Utils/JsonUtil.cs b/MasstransferCommon/Utils/JsonUtil.cs
index 37b453c..945d980 100644
--- a/MasstransferCommon/Utils/JsonUtil.cs
+++ b/MasstransferCommon/Utils/JsonUtil.cs
@@ -40,6 +40,18 @@ public class JsonUtil
         }
     }
 
+    public static object? FromJson(Type type, string json)
+    {
+        try
+        {
+            return JsonConvert.DeserializeObject(json, type);
+        }
+        catch (Exception e)
+        {
+            throw new ArgumentException($" 无效的json 字符串 {json} ");
+        }
+    }
+
     public static T FromJsonOrDefault<T>(string json)
     {
         try
diff --git a/MasstransferExporter/Stat/HeartbeatExporter.cs b/MasstransferExporter/Stat/HeartbeatExporter.cs
new file mode 100644
index 0000000..d7ceaef
--- /dev/null
+++ b/MasstransferExporter/Stat/HeartbeatExporter.cs
@@ -0,0 +1,22 @@
+using MasstransferCommon.Scheduler;
+
+namespace MasstransferExporter.Stat;
+
+/// <summary>
+/// 心跳
+/// </summary>
+public class HeartbeatExporter
+{
+    /// <summary>
+    /// 启动心跳线程
+    /// </summary>
+    public static void StartHeartBeat()
+    {
+        JobScheduler.AddTask("HeartbeatExporter#StartHeartBeat", HeartBeat, 10000);
+    }
+
+    private static void HeartBeat()
+    {
+        
+    }
+}
\ No newline at end of file
diff --git a/MasstransferInfrastructure/MasstransferInfrastructure.csproj b/MasstransferInfrastructure/MasstransferInfrastructure.csproj
index a3dd507..52f38e2 100644
--- a/MasstransferInfrastructure/MasstransferInfrastructure.csproj
+++ b/MasstransferInfrastructure/MasstransferInfrastructure.csproj
@@ -15,6 +15,7 @@
 
     <ItemGroup>
         <PackageReference Include="Masuit.Tools.Core" Version="2024.3.4" />
+        <PackageReference Include="Minio" Version="6.0.2" />
         <PackageReference Include="MQTTnet" Version="4.3.6.1152"/>
         <PackageReference Include="Serilog" Version="4.0.0"/>
         <PackageReference Include="Microsoft.Data.Sqlite" Version="9.0.0-preview.2.24128.4"/>
diff --git a/MasstransferInfrastructure/Minio/MinioHelper.cs b/MasstransferInfrastructure/Minio/MinioHelper.cs
new file mode 100644
index 0000000..7f0a86e
--- /dev/null
+++ b/MasstransferInfrastructure/Minio/MinioHelper.cs
@@ -0,0 +1,88 @@
+using Minio;
+using Minio.DataModel.Args;
+
+namespace MasstransferCommunicate.Minio;
+
+/// <summary>
+/// Minio工具类
+/// </summary>
+public class MinioHelper
+{
+    private readonly IMinioClient _client;
+
+    public MinioHelper(string endpoint, string accessKey, string secretKey)
+    {
+        _client = new MinioClient()
+            .WithEndpoint(endpoint)
+            .WithCredentials(accessKey, secretKey)
+            .Build();
+    }
+
+    /// <summary>
+    /// 判断bucket 是否存在
+    /// </summary>
+    /// <param name="bucketName"></param>
+    /// <returns></returns>
+    public async Task<bool> BucketExistsAsync(string bucketName)
+    {
+        return await _client.BucketExistsAsync(new BucketExistsArgs().WithBucket(bucketName));
+    }
+
+    /// <summary>
+    /// 上传文件
+    /// </summary>
+    /// <param name="bucketName"></param>
+    /// <param name="fileName"></param>
+    /// <param name="filePath"></param>
+    /// <returns></returns>
+    public async Task<string> UploadFileAsync(string bucketName, string fileName, string filePath)
+    {
+        if (!await BucketExistsAsync(bucketName))
+        {
+            await _client.MakeBucketAsync(new MakeBucketArgs().WithBucket(bucketName));
+        }
+
+        await _client.PutObjectAsync(new PutObjectArgs()
+            .WithBucket(bucketName)
+            .WithObject(fileName)
+            .WithFileName(filePath));
+
+        return $"{bucketName}/{fileName}";
+    }
+
+    /// <summary>
+    /// 下载文件
+    /// </summary>
+    /// <param name="bucketName"></param>
+    /// <param name="fileName"></param>
+    /// <param name="filePath"></param>
+    public async Task DownloadFileAsync(string bucketName, string fileName, string filePath)
+    {
+        if (!await BucketExistsAsync(bucketName))
+        {
+            await _client.MakeBucketAsync(new MakeBucketArgs().WithBucket(bucketName));
+        }
+
+        await _client.GetObjectAsync(new GetObjectArgs()
+            .WithBucket(bucketName)
+            .WithObject(fileName)
+            .WithFile(filePath));
+    }
+
+    /// <summary>
+    /// 删除文件
+    /// </summary>
+    /// <param name="bucketName"></param>
+    /// <param name="fileName"></param>
+    public async Task DeleteFileAsync(string bucketName, string fileName)
+    {
+        if (!await BucketExistsAsync(bucketName))
+        {
+            await _client.MakeBucketAsync(new MakeBucketArgs().WithBucket(bucketName));
+        }
+
+        await _client.RemoveObjectAsync(new RemoveObjectArgs()
+            .WithBucket(bucketName)
+            .WithObject(fileName));
+    }
+}
\ No newline at end of file
diff --git a/MasstransferInfrastructure/Mqtt/Client/MessageQueueHelper.cs b/MasstransferInfrastructure/Mqtt/Client/MessageQueueHelper.cs
index 9f42de8..f645ff5 100644
--- a/MasstransferInfrastructure/Mqtt/Client/MessageQueueHelper.cs
+++ b/MasstransferInfrastructure/Mqtt/Client/MessageQueueHelper.cs
@@ -1,4 +1,5 @@
-using MasstransferCommon.Utils;
+using System.Reflection;
+using MasstransferCommon.Utils;
 using MasstransferInfrastructure.Mqtt.Model;
 using MQTTnet;
 using MQTTnet.Client;
@@ -7,11 +8,10 @@ using Serilog;
 
 namespace MasstransferInfrastructure.Mqtt.Client;
 
-public class MessageQueueHelper<T>
+public class MessageQueueHelper
 {
-    private static readonly Dictionary<string, List<Action<string, T>>> Subscribers = new();
+    private static readonly Dictionary<string, List<Delegate>> Subscribers = new();
 
-    // ReSharper disable once StaticMemberInGenericType
     private static readonly MqttClient Client = new();
 
 
@@ -41,7 +41,7 @@ public class MessageQueueHelper<T>
     /// <param name="topic"></param>
     /// <param name="delegate"></param>
     /// <param name="qos"></param>
-    public static async Task<bool> Subscribe(string topic, Action<string, T> @delegate,
+    public static async Task<bool> Subscribe(string topic, Delegate @delegate,
         MqttQualityOfServiceLevel qos = MqttQualityOfServiceLevel.AtMostOnce)
     {
         if (!Subscribers.ContainsKey(topic))
@@ -84,8 +84,12 @@ public class MessageQueueHelper<T>
         {
             try
             {
+                var methodInfo = subscriber.Method;
+                var parameters = methodInfo.GetParameters();
+                if (parameters.Length != 2) continue;
+                var type = parameters[1].ParameterType;
                 // 通知订阅者
-                subscriber(topic, JsonUtil.FromJson<T>(message));
+                subscriber.DynamicInvoke(topic, JsonUtil.FromJson(type, message));
             }
             catch (Exception exception)
             {