Add pubsub node forwarder

CloudSync.sln

@@ -5,6 +5,11 @@ VisualStudioVersion = 17.1.32407.343
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "CloudSync", "CloudSync\CloudSync.csproj", "{E3947976-AF75-45D7-AAE8-D8878E648AD7}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Docs", "Docs", "{1D979090-62F4-4085-89BC-E74466EC6508}"
+	ProjectSection(SolutionItems) = preProject
+		README.md = README.md
+	EndProjectSection
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU

CloudSync/Config/GetKetQuaKCBPubSub.json

@@ -1,5 +1,5 @@
 {
-  "Enabled": true,
+  "Enabled": false,
   "ApiHisKqKCBUrl": "",
   "ApiPostKetQuaKCB": "",
   "ApiHisToken": "",

CloudSync/Config/NodeForwarderPubSub.json

@@ -0,0 +1,15 @@
+{
+  "Enabled": true,
+  "CloudGetUrl": "",
+  "CloudToken": "",
+  "NodeURL": "",
+  "ForwardAuthKey": "",
+  "Requests": [
+    {
+      "Topic": "emr/hs/pending",
+      "ExecuteApi": "his-emr_list?lane=hs&msgToken={token}",
+      "ForwardUrl": "https://httpbin.org/post",
+      "ForwardAuthKey": ""
+    }
+  ]
+}

CloudSync/Process/Dto/TopicSubRequest.cs

@@ -0,0 +1,10 @@
+namespace CloudSync.Process.Dto
+{
+    public class TopicSubRequest
+    {
+        public string Topic { get; set; }
+        public string ExecuteApi { get; set; }
+        public string ForwardUrl { get; set; }
+        public string ForwardAuthKey { get; set; }
+    }
+}

CloudSync/Process/Dto/TopicSubResult.cs

@@ -0,0 +1,9 @@
+namespace CloudSync.Process.Dto
+{
+    public class TopicSubResult
+    {
+        public TopicSubRequest Request { get; set; }
+        public object Data { get; set; }
+        public string Token { get; set; }
+    }
+}

CloudSync/Process/GetKetQuaKCBPubSub.cs

@@ -39,7 +39,7 @@
                 return Task.FromResult(true);
             }, (subErr) =>
             {
                 if (subErr is Exception)
                 {
                     logger.Error("Please check subError!", subErr);
                 }
@@ -159,5 +159,12 @@
         public string XMLData { get; set; }
         public string MaLanKham { get; set; }
         public string CCCD { get; set; }
+        public string FileDataURL { get; set; }
+        /// <summary>
+        /// 0: Kho BH
+        /// 1: Kho HSBA
+        /// 2: Kho GiayChuyenTuyen
+        /// </summary>
+        public int Type { get; set; }
     }
 }

CloudSync/Process/NodeForwarderPubSub.cs

@@ -0,0 +1,123 @@
+using CQDT.CloudClient;
+using HSDT.Common.Exceptions;
+using HSDT.Common;
+using Newtonsoft.Json;
+using RestSharp;
+using System;
+using HSDT.AutoSync.Process;
+using System.Threading.Tasks;
+using HSDT.Common.Helper;
+using CloudSync.Process.Dto;
+
+namespace CloudSync.Process
+{
+    public class NodeForwarderPubSub : QueueJobProcess<TopicSubResult>
+    {
+        private string latestMessageToken = string.Empty;
+
+        /// <summary>
+        /// - Khởi tạo sub, lắng nghe các sự kiện Pub/sub.
+        /// - Chờ nhận tín hiệu tử cloud.
+        /// </summary>
+        protected override void PostInitialize()
+        {
+            var authToken = Config.FindValue("CloudToken");
+            //var subChannel = Config.FindValue("Channel", "phieukhamkq/pending");
+            var vNodeURL = Config.FindValue("NodeURL", "");
+            var infoToken = authToken?.Substring(0, 6) ?? "NULL";
+            var subRequests = Config.FindValue("Requests", "");
+            if (subRequests.IsNullOrEmpty())
+            {
+                logger.Error("Pub/sub Requests chưa được cấu hình!");
+                return;
+            }
+            if (authToken is null || !PubSubHelper.Initialize(authToken, vNodeURL).Wait(10000))
+            {
+                logger.Error($"Cannot init Pubsub => Please verify CloudToken, token={infoToken}...");
+                return;
+            }
+            logger.Info($"PubSubHelper.Initialize ok!: token={infoToken}...");
+            var vSubRequests = JsonConvert.DeserializeObject<TopicSubRequest[]>(subRequests);
+            foreach (var item in vSubRequests)
+            {
+                var topic = item.Topic;
+                logger.Info($"Starting sub topic: {item.Topic}, {item.ExecuteApi}, {item.ForwardUrl}");
+                PubSubHelper.Subcribe(topic, async (subTopic, token) =>
+                {
+                    this.latestMessageToken = token;
+                    // $"{item.ExecuteApi}&msgToken={token}";
+                    var apiMethodUrl = item.ExecuteApi.Interpolate(new { token });
+                    logger.Info($"0.0. Raised event, subTopic={subTopic}, apiMethodUrl={apiMethodUrl}...");
+                    var vGetData = await apiMethodUrl.GetAsJson<object[]>(authToken, vNodeURL);
+                    if (vGetData.Length > 0)
+                    {
+                        var vResult = new TopicSubResult();
+                        vResult.Request = item;
+                        vResult.Token = token;
+                        vResult.Data = vGetData;
+                        this.Enqueue(vResult);
+                        // Alert scanner!
+                        this.Alert();
+                    }
+                    return true;
+                }, (subErr) =>
+                {
+                    if (subErr is Exception)
+                    {
+                        logger.Error("Please check subError!", subErr);
+                    }
+                    else
+                    {
+                        logger.Info($"Listening... subErr={subErr},subTopic={topic}");
+                    }
+                    return true;
+                });
+            }
+            logger.Info($"Done sub topic length: {vSubRequests.Length}!");
+        }
+
+        /// <summary>
+        /// - Kéo các kết quả phiếu đón tiếp từ Cloud 
+        /// - Đẩy vào queue chờ xử lý.
+        /// </summary>
+        protected override void Scan()
+        {
+            // TODO: something
+            logger.Info("Process still alive!");
+            this.Notify();
+        }
+
+        /// <summary>
+        /// - Lấy dữ liệu tuần tự từ Queue.
+        /// - Gửi dữ liệu về HIS.
+        /// </summary>
+        /// <param name="data"></param>
+        /// <exception cref="HSCoreException"></exception>
+        protected override void Process(TopicSubResult data)
+        {
+            if (data is null)
+            {
+                return;
+            }
+            var vForwardAuthKey = data.Request.ForwardAuthKey ?? Config.FindValue("ForwardAuthKey");
+            var client = new RestClient(data.Request.ForwardUrl);
+            // Gửi kết quả về HIS.
+            var vRequest = new RestRequest(data.Request.ForwardUrl, Method.Post)
+                .AddJsonBody(data.Data.ToString());
+            if (vForwardAuthKey.IsNotNullOrEmpty())
+            {
+                vRequest.AddHeader("Authorization", vForwardAuthKey);
+            }
+            // Send request
+            var vResponse = client.ExecuteAsync<string>(vRequest).Result;
+            if (vResponse.IsSuccessStatusCode)
+            {
+                logger.Info($"2. Gửi kết quả thành công: url={data.Request.ForwardUrl},status={vResponse.StatusCode},result={vResponse.Content}");
+            }
+            else
+            {
+                logger.Error($"2. Gửi kết quả thất bại: url={data.Request.ForwardUrl},status={vResponse.StatusCode},error={vResponse.Content}");
+            }
+        }
+    }
+}

README.md

@@ -25,6 +25,14 @@
 - `Channel`: Tên channel nhận dữ liệu phiếu đón tiếp (mặc định do Cloud cung cấp)
 - `Lane`: Tên kênh lane nhận dữ liệu, khi cấu hình nhiều kênh (mặc định để trống)
 
+Các cấu hình sẽ ưu tiên theo thứ tự:
+
+1. Environment Variables
+2. Process config: ./Config/GetPhieuDonTiepPubSub.json
+3. App config: ./appsettings.json & ./appsettings.Production.json
+
+Trong quá trình phát triển, cấu hình có thể được đặt trong file: `appsettings.Development.json`
+
 
 ## - Build dự án:
 ### + Windows: