From f575a1d2465f9f4f1be4d654e43d624ba2984fbc Mon Sep 17 00:00:00 2001
From: engigu <sayheya@qq.com>
Date: Wed, 3 Jan 2024 19:55:38 +0800
Subject: [PATCH] feat: add send message async

---
 main.go                                       | 16 ++++++-
 pkg/constant/constant.go                      |  3 ++
 routers/api/v1/send_message.go                | 19 ++++++---
 service/send_message_service/consumer.go      | 42 +++++++++++++++++++
 service/send_message_service/send_message.go  |  1 +
 web/src/util/viewApi.js                       |  2 +-
 .../sendTasks/view/viewTaskAPIPopUp.vue       |  2 +-
 7 files changed, 76 insertions(+), 9 deletions(-)
 create mode 100644 service/send_message_service/consumer.go

diff --git a/main.go b/main.go
index e6ed65a..ec5f2ab 100644
--- a/main.go
+++ b/main.go
@@ -7,7 +7,9 @@ import (
 	"message-nest/pkg/util"
 	"message-nest/service/cron_service"
 	"message-nest/service/env_service"
+	"message-nest/service/send_message_service"
 	"net/http"
+	"sync"
 
 	"github.com/gin-gonic/gin"
 
@@ -20,6 +22,8 @@ import (
 var (
 	//go:embed web/dist/*
 	f embed.FS
+
+	wg sync.WaitGroup
 )
 
 func init() {
@@ -32,7 +36,7 @@ func init() {
 	cron_service.Setup()
 }
 
-func main() {
+func GinServerUp() {
 	gin.SetMode(setting.ServerSetting.RunMode)
 
 	routersInit := routers.InitRouter(f)
@@ -56,3 +60,13 @@ func main() {
 		logging.Logger.Error("Server err: ", err)
 	}
 }
+
+func main() {
+	wg.Add(1)
+
+	go GinServerUp()
+	go send_message_service.MessageConsumer(&wg)
+
+	wg.Wait()
+	fmt.Println("Server exit...")
+}
diff --git a/pkg/constant/constant.go b/pkg/constant/constant.go
index abd7a8f..07aa3d6 100644
--- a/pkg/constant/constant.go
+++ b/pkg/constant/constant.go
@@ -26,3 +26,6 @@ var LogsCleanDefaultValueMap = map[string]string{
 }
 
 const AboutSectionName = "about"
+
+// 限制goroutine的最大数量
+var MaxSemaphore = make(chan struct{}, 2048)
diff --git a/routers/api/v1/send_message.go b/routers/api/v1/send_message.go
index 997ba4e..91a11c7 100644
--- a/routers/api/v1/send_message.go
+++ b/routers/api/v1/send_message.go
@@ -13,6 +13,7 @@ type SendMessageReq struct {
 	Text     string `json:"text" validate:"required" label:"文本内容"`
 	HTML     string `json:"html"  label:"html内容"`
 	MarkDown string `json:"markdown" label:"markdown内容"`
+	Mode     string `json:"mode" label:"是否异步发送"`
 }
 
 // DoSendMassage 外部调用发信接口
@@ -34,11 +35,17 @@ func DoSendMassage(c *gin.Context) {
 		HTML:     req.HTML,
 		MarkDown: req.MarkDown,
 	}
-	err := msgService.Send()
-	if err != "" {
-		appG.CResponse(http.StatusBadRequest, "发送失败！", nil)
-		return
+	if req.Mode != "async" {
+		// 同步发送
+		err := msgService.Send()
+		if err != "" {
+			appG.CResponse(http.StatusBadRequest, "发送失败！", nil)
+			return
+		}
+		appG.CResponse(http.StatusOK, "发送成功！", nil)
+	} else {
+		// 异步发送
+		send_message_service.Buffer <- msgService
+		appG.CResponse(http.StatusOK, "提交成功！", nil)
 	}
-
-	appG.CResponse(http.StatusOK, "发送成功！", nil)
 }
diff --git a/service/send_message_service/consumer.go b/service/send_message_service/consumer.go
new file mode 100644
index 0000000..e862904
--- /dev/null
+++ b/service/send_message_service/consumer.go
@@ -0,0 +1,42 @@
+package send_message_service
+
+import (
+	"message-nest/pkg/constant"
+	"message-nest/pkg/logging"
+	"sync"
+)
+
+var maxBufferSize = 10
+var Buffer = make(chan SendMessageService, maxBufferSize)
+
+func DoSendTask(task SendMessageService, wg *sync.WaitGroup) {
+	defer wg.Done()
+	defer func() {
+		if r := recover(); r != nil {
+			logging.Logger.Error("DoSendTask: Recovered from panic:", r)
+		}
+	}()
+
+	constant.MaxSemaphore <- struct{}{}
+	defer func() {
+		<-constant.MaxSemaphore
+	}()
+
+	go task.Send()
+}
+
+func MessageConsumer(wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	for {
+		task, ok := <-Buffer
+		if !ok {
+			logging.Logger.Error("MessageConsumer: Channel closed. Exiting.")
+			return
+		}
+
+		wg.Add(1)
+		go DoSendTask(task, wg)
+	}
+
+}
diff --git a/service/send_message_service/send_message.go b/service/send_message_service/send_message.go
index b9d55d4..527b81a 100644
--- a/service/send_message_service/send_message.go
+++ b/service/send_message_service/send_message.go
@@ -37,6 +37,7 @@ func (sm *SendMessageService) LogsAndStatusMark(errStr string, status int) {
 	if status == SendFail {
 		sm.Status = SendFail
 	}
+	logging.Logger.Info(fmt.Sprintf("%s, 状态：%d", errStr, status))
 }
 
 // Send 发送一个消息任务的所有实例
diff --git a/web/src/util/viewApi.js b/web/src/util/viewApi.js
index 02905d9..4631adf 100644
--- a/web/src/util/viewApi.js
+++ b/web/src/util/viewApi.js
@@ -8,7 +8,7 @@ class ApiStrGenerate {
 
 
     static getCurlString(task_id, options) {
-        let data = { task_id: task_id };
+        let data = { task_id: task_id, mode: "async" };
         data.text = 'Hello World!';
         if (options.html) {
             data.html = '<h1> Hello World! </h1>';
diff --git a/web/src/views/tabsTools/sendTasks/view/viewTaskAPIPopUp.vue b/web/src/views/tabsTools/sendTasks/view/viewTaskAPIPopUp.vue
index a2fa1d3..daab544 100644
--- a/web/src/views/tabsTools/sendTasks/view/viewTaskAPIPopUp.vue
+++ b/web/src/views/tabsTools/sendTasks/view/viewTaskAPIPopUp.vue
@@ -10,7 +10,7 @@
           <br />
           发送的消息会优先现在相应的类型消息进行发送，如果没有，将使用传的text消息进行发送          
           <br />
-          *text节点必传
+          ** text节点必传，指定mode=async将异步发送，否则同步发送
         </template>
         <el-icon>
           <QuestionFilled />