eta_api/services/websocket_msg.go

package services

import (
	"eta/eta_api/models"
	"eta/eta_api/services/data"
	"eta/eta_api/utils"
	"fmt"

	"github.com/gorilla/websocket"
)

func DealWebSocketMsg(conn *websocket.Conn, adminId int) {
	DealEdbInspectionMessage(conn, adminId)
}

// 处理巡检消息
func DealEdbInspectionMessage(conn *websocket.Conn, adminId int) {
	cacheKey := fmt.Sprintf("%s%d", utils.CACHE_EDB_INSPECTION_MESSAGE, adminId)
	for {
		utils.Rc.Brpop(cacheKey, func(b []byte) {
			messageList, err := data.GetHistoryInspectionMessages(adminId)
			if err != nil {
				utils.FileLog.Error("获取巡检信息历史失败,err:%s, adminId:%d", err.Error(), adminId)
				return
			}
			success := make(chan int64, 10)
			defer close(success)
			go func() {
				defer close(success)
				for i, msg := range messageList {
					if i == 0 {
						// 多条消息仅发送最新一条
						respData, err := data.SendInspectionMessages(adminId, msg)
						if err != nil {
							utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
						} else {
							resp := models.WebsocketMessageResponse{
								MessageType: 1,
								Data: respData,
							}
							err = conn.WriteJSON(resp)
							if err != nil {
								utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
							} else {
								utils.FileLog.Info("巡检信息发送成功,adminId:%d, messageId:%d", adminId, msg.MessageId)
								success <- msg.MessageId
							}
						}
					} else {
						success <- msg.MessageId
					}
				}
			}()
			go func() {
				readList := make([]int64, 0)
				for {
					msgId, ok := <-success
					if !ok {
						break
					}
					readList = append(readList, msgId)
				}
				_, err = data.ReadEdbInspectionMessageList(readList, adminId)
				if err != nil {
					utils.FileLog.Error("巡检信息已读失败,err:%s, adminId:%d", err.Error(), adminId)
				}
			}()
			})
	}
	
}