eta_api/services/websocket_msg.go

package services

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

	"context"	

	"github.com/gorilla/websocket"
)

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

// 处理巡检消息
func DealEdbInspectionMessage(conn *websocket.Conn, adminId int) {
	// 创建上下文用于控制 goroutine 生命周期
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// 创建互斥锁保护 WebSocket 写操作
	var wsWriteMutex sync.Mutex

	// // 监控 goroutine 数量
	// go func() {
	// 	ticker := time.NewTicker(time.Minute)
	// 	defer ticker.Stop()
	// 	for {
	// 		select {
	// 		case <-ticker.C:
	// 			utils.FileLog.Info("Current goroutine count: %d", runtime.NumGoroutine())
	// 		case <-ctx.Done():
	// 			return
	// 		}
	// 	}
	// }()

	cacheKey := fmt.Sprintf("%s%d", utils.CACHE_EDB_INSPECTION_MESSAGE, adminId)

	// 设置连接关闭处理器
	conn.SetCloseHandler(func(code int, text string) error {
		utils.FileLog.Info("长连接关闭, adminId:%d", adminId)
		cancel()
		return nil
	})

	// 添加错误恢复机制
	defer func() {
		if r := recover(); r != nil {
			utils.FileLog.Error("WebSocket handler recovered from panic: %v", r)
			// 清理资源
			cancel()
		}
	}()

	for {
		select {
		case <-ctx.Done():
			return
		default:
			// 使用带超时的 Redis 操作
			err := utils.Rc.BrpopWithTimeout(cacheKey, 30*time.Second, func(b []byte) {
				utils.FileLog.Info("收到巡检信息开始处理, adminId:%d", adminId)

				messageList, err := data.GetHistoryInspectionMessages(adminId)
				if err != nil {
					utils.FileLog.Error("获取巡检信息历史失败,err:%s, adminId:%d", err.Error(), adminId)
					return
				}
				if len(messageList) == 0 {
					utils.FileLog.Info("巡检信息历史为空, adminId:%d", adminId)
					return
				}
				readList := make([]int64, 0)
				// 消息发送 goroutine

				for i, msg := range messageList {
					select {
					case <-ctx.Done():
							return
					default:
						if i == 0 {
							respData, err := data.SendInspectionMessages(adminId, msg)
							if err != nil {
								utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
								continue
							}

							resp := models.WebsocketMessageResponse{
								MessageType: 1,
								Data:       respData,
							}

							// 使用互斥锁保护 WebSocket 写操作
							wsWriteMutex.Lock()
							err = conn.WriteJSON(resp)
							if err != nil {
								utils.FileLog.Error("巡检信息写通道被锁，消息发送失败,err:%s, adminId:%d", err.Error(), adminId)
								wsWriteMutex.Unlock()
								continue
							}
							wsWriteMutex.Unlock()

							if err != nil {
								utils.FileLog.Error("巡检信息发送失败,err:%s, adminId:%d", err.Error(), adminId)
								continue
							}

							utils.FileLog.Info("巡检信息发送成功,adminId:%d, messageId:%d", adminId, msg.MessageId)
							readList = append(readList, msg.MessageId)
						} else {
							readList = append(readList, msg.MessageId)
						}
					}
				}

				if len(readList) > 0 {
					_, err = data.ReadEdbInspectionMessageList(readList, adminId)
					if err != nil {
						utils.FileLog.Error("巡检信息已读失败,err:%s, adminId:%d", err.Error(), adminId)
					}
				}
			})

			if err != nil {
				utils.FileLog.Error("Redis operation failed: %v", err)
				continue
			}else{
				utils.FileLog.Info("巡检信息处理完成, adminId:%d", adminId)
			}
		}
	}
}