eta_api/utils/ws/session_manager.go

package ws

import (
	"errors"
	"fmt"
	"github.com/gorilla/websocket"
	"sync"
	"time"
)

const (
	maxMessageSize   = 1024 * 1024 // 1MB
	basePingInterval = 30 * time.Second
	maxPingInterval  = 120 * time.Second
	minPingInterval  = 15 * time.Second
)

type ConnectionManager struct {
	Sessions  sync.Map
	heartbeat *HeartbeatManager
}

var (
	manager = &ConnectionManager{
		heartbeat: GetHeartbeatManager(),
	}
)

func Manager() *ConnectionManager {
	return manager
}

// 消息处理核心逻辑
func (manager *ConnectionManager) HandleMessage(userID int, sessionID string, message []byte) error {
	if !Allow(userID, QA_LIMITER) {
		return errors.New("request too frequent")
	}
	session, exists := manager.GetSession(sessionID)
	if !exists {
		return errors.New("session not found")
	}

	// 处理业务逻辑
	session.History = append(session.History, string(message))
	response := "Processed: " + string(message)

	// 更新最后活跃时间
	session.LastActive = time.Now()

	// 发送响应
	return session.Conn.WriteMessage(websocket.TextMessage, []byte(response))
}

// 心跳管理
func (manager *ConnectionManager) StartHeartbeat() {
	ticker := time.NewTicker(basePingInterval)
	defer ticker.Stop()
	for range ticker.C {
		manager.checkSessions()
	}
}

func (manager *ConnectionManager) checkSessions() {
	manager.Sessions.Range(func(key, value interface{}) bool {
		session := value.(*Session)
		if time.Since(session.LastActive) > 2*basePingInterval {
			session.Conn.Close()
			manager.Sessions.Delete(key)
		} else {
			_ = session.Latency.SendPing(session.Conn)
		}
		return true
	})
}

// AddSession Add 添加一个新的会话
func (manager *ConnectionManager) AddSession(session *Session) {
	manager.Sessions.Store(session.ID, session)
	manager.heartbeat.Register(session)
}
func (manager *ConnectionManager) GetSessionId(userId int, sessionId string) (sessionID string) {
	return fmt.Sprintf("%d_%s", userId, sessionId)
}

// RemoveSession Remove 移除一个会话
func (manager *ConnectionManager) RemoveSession(sessionCode string) {
	if data, ok := manager.Sessions.LoadAndDelete(sessionCode); ok {
		session := data.(*Session)
		close(session.CloseChan)
		_ = session.Conn.Close()
	}
}

// GetSession 获取一个会话
func (manager *ConnectionManager) GetSession(sessionCode string) (session *Session, ok bool) {
	if data, ok := manager.Sessions.Load(sessionCode); ok {
		session = data.(*Session)
	}
	return
}