nn.daxia.dev/online/online.go

package online

import (
	"fmt"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	"github.com/patrickmn/go-cache"
	"github.com/sirupsen/logrus"
	"gogs.daxia.dev/huanan/pkg.daxia.dev/event2"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/reflect/protoreflect"
	"nn.daxia.dev/gameproto"
	"nn.daxia.dev/model"
)

type Msg struct {
	RoomID     uint32
	Offset     int
	NotifyType uint32
	Data       string
}

type EventMsg struct {
	Msg Msg
}

type EventConnect struct {
	UserModel *model.User
	Conn      *websocket.Conn
}
type EventDisconnect struct {
	UserModel *model.User
	Conn      *websocket.Conn
}
type EventPing struct {
	UserModel *model.User
	Conn      *websocket.Conn
}

type OnlineUser struct {
	UserID uint

	RoomID uint32

	ChairID uint32

	Conn *websocket.Conn

	Locker sync.Mutex

	IsRunning bool

	Ready bool

	Event chan bool

	MsgList []Msg

	MsgIndex int
}

func NewOnlineUser(userID uint, roomID, chairID uint32, conn *websocket.Conn) *OnlineUser {
	return &OnlineUser{
		UserID:    userID,
		RoomID:    roomID,
		ChairID:   chairID,
		Conn:      conn,
		Locker:    sync.Mutex{},
		IsRunning: false,
		Ready:     false,
		Event:     make(chan bool),
		MsgList:   make([]Msg, 0),
		MsgIndex:  0,
	}
}

func (p *OnlineUser) Start() {
	p.IsRunning = true

	//监听消息
	evnetUser := model.EventNotifyMsg + fmt.Sprintf("user_%d", p.UserID)

	event2.On[EventMsg](evnetUser, event2.ListenerFunc[EventMsg](func(e event2.Event[EventMsg]) error {
		p.Locker.Lock()
		defer p.Locker.Unlock()

		if !p.IsRunning {
			return nil
		}

		msg := e.Data().Msg

		p.MsgIndex = p.MsgIndex + 1
		msg.Offset = p.MsgIndex

		p.MsgList = append(p.MsgList, msg)

		logrus.Debugf("event ready send: %v", evnetUser)

		go func() {
			defer func() {
				recover()
			}()
			p.Event <- true
		}()

		return nil
	}))

	//退出则清理一下
	defer func() {
		//cancel event listener
		event2.On[EventMsg](evnetUser, event2.ListenerFunc[EventMsg](func(e event2.Event[EventMsg]) error {
			return nil
		}))

		p.Stop()
	}()

	for {
		if !p.IsRunning {
			break
		}

		<-p.Event

		if !p.Ready {
			continue
		}

		//遍历，发送通知
		sendMsgList := func() []Msg {
			p.Locker.Lock()
			defer p.Locker.Unlock()

			msgTmp := make([]Msg, len(p.MsgList))

			copy(msgTmp, p.MsgList)
			p.MsgList = make([]Msg, 0)

			return msgTmp
		}()

		for _, item := range sendMsgList {
			err := SendMsg(p.Conn, EmitData{
				UserID:     p.UserID,
				NotifyType: gameproto.NotifyTypeEnum(item.NotifyType),
				Encrypt:    false,
				Data:       item.Data,
			})

			if err != nil {
				logrus.Error(err)
				return
			}
		}

		fmt.Println("running")
	}
}

func (p *OnlineUser) SetReady(msg Msg) {
	if !p.IsRunning {
		return
	}

	p.Locker.Lock()
	defer p.Locker.Unlock()

	p.Ready = true
	p.MsgIndex = 1
	msg.Offset = 1
	p.MsgList = make([]Msg, 0)

	p.MsgList = append(p.MsgList, msg)

	go func() {
		defer func() {
			recover()
		}()
		p.Event <- true
	}()
}

func (p *OnlineUser) Stop() {
	if !p.IsRunning {
		return
	}

	p.Locker.Lock()
	defer p.Locker.Unlock()

	p.IsRunning = false

	go func() {
		p.Event <- true
		close(p.Event)
	}()

	p.Conn.Close()
}

var OnlineUserCache = cache.New(time.Minute*3, 10*time.Minute)
var locker = sync.Mutex{}

func Init() {
	//删除的时候，取消
	OnlineUserCache.OnEvicted(func(key string, value interface{}) {
		onlineUser := value.(*OnlineUser)
		onlineUser.Stop()
	})

	event2.On[EventConnect](model.EventConnect, event2.ListenerFunc[EventConnect](func(e event2.Event[EventConnect]) error {
		userModel := e.Data().UserModel
		conn := e.Data().Conn

		Active(userModel.ID, conn)

		return nil
	}))

	event2.On[EventPing](model.EventPing, event2.ListenerFunc[EventPing](func(e event2.Event[EventPing]) error {
		userModel := e.Data().UserModel
		conn := e.Data().Conn

		Active(userModel.ID, conn)

		return nil
	}))

	event2.On[EventDisconnect](model.EventDisconnect, event2.ListenerFunc[EventDisconnect](func(e event2.Event[EventDisconnect]) error {
		userModel := e.Data().UserModel
		conn := e.Data().Conn

		UnActive(userModel.ID, conn)

		return nil
	}))
}

func Active(userID uint, conn *websocket.Conn) {
	//logrus.Infof("active user id:%d", userID)

	locker.Lock()
	defer locker.Unlock()

	var onlineUser *OnlineUser

	userKey := GetKey(userID)
	user, exists := OnlineUserCache.Get(userKey)
	if !exists {
		onlineUser = &OnlineUser{
			UserID:    userID,
			RoomID:    0,
			ChairID:   0,
			Conn:      conn,
			Locker:    sync.Mutex{},
			IsRunning: false,
			Event:     make(chan bool),
			MsgList:   make([]Msg, 0),
			MsgIndex:  0,
		}

		go onlineUser.Start()

		OnlineUserCache.Set(userKey, onlineUser, time.Minute*3)
	} else {
		currConn := user.(*OnlineUser).Conn
		if currConn != conn {
			//TODO:发送掉线信息
			data, err := proto.Marshal(&gameproto.Disconnect{
				UserID: uint32(userID),
			})

			if err != nil {
				logrus.Error(err)
				return
			}

			SendMsg(currConn, EmitData{
				UserID:     userID,
				NotifyType: gameproto.NotifyTypeEnum_NotifyTypeDisconnect,
				Encrypt:    false,
				Data:       string(data),
			})

			currConn.Close()

			user.(*OnlineUser).Conn = conn
		}

		OnlineUserCache.Set(userKey, user, time.Minute*3)
	}
}

func SetReady(userID uint, notifyType gameproto.NotifyTypeEnum, dataSend protoreflect.ProtoMessage) {
	locker.Lock()
	defer locker.Unlock()

	userKey := GetKey(userID)
	userObj, exists := OnlineUserCache.Get(userKey)
	if !exists {
		logrus.Error("用户不存在:", userID)
		return
	}

	data, err := proto.Marshal(dataSend)

	if err != nil {
		logrus.Error(err)
		return
	}

	user := userObj.(*OnlineUser)
	user.SetReady(Msg{
		NotifyType: uint32(notifyType),
		Data:       string(data),
	})
}

func IsActive(userID uint) bool {
	locker.Lock()
	defer locker.Unlock()

	userKey := GetKey(userID)
	_, exists := OnlineUserCache.Get(userKey)
	return exists
}

func UnActive(userID uint, conn *websocket.Conn) {
	locker.Lock()
	defer locker.Unlock()

	userKey := GetKey(userID)
	user, exists := OnlineUserCache.Get(userKey)
	if !exists {
		return
	}

	currConn := user.(*OnlineUser).Conn
	if currConn != conn {
		return
	}

	OnlineUserCache.Delete(userKey)
}

func GetUserByRoom(roomID uint32) []uint32 {
	onlineUserIDList := make([]uint32, 0)

	roomModel := model.Room{}
	userList, err := roomModel.GetRoomUserList(roomID)
	if err != nil {
		logrus.Error(err)
		return onlineUserIDList
	}

	return userList
}

func GetUserByNotInRoom(roomID uint32) []uint32 {
	onlineUserIDList := make([]uint32, 0)

	locker.Lock()
	defer locker.Unlock()

	items := OnlineUserCache.Items()
	for _, item := range items {
		onlineUser := item.Object.(*OnlineUser)
		if onlineUser.RoomID == roomID {
			continue
		}

		onlineUserIDList = append(onlineUserIDList, uint32(onlineUser.UserID))
	}

	return onlineUserIDList
}

func GetAllUser() []uint32 {
	onlineUserIDList := make([]uint32, 0)

	locker.Lock()
	defer locker.Unlock()

	items := OnlineUserCache.Items()
	for _, item := range items {
		onlineUser := item.Object.(*OnlineUser)
		onlineUserIDList = append(onlineUserIDList, uint32(onlineUser.UserID))
	}

	return onlineUserIDList
}

func GetKey(userID uint) string {
	return fmt.Sprintf("online_user_%d", userID)
}