s_cherry_e/sr/internal/sessions/actor.go

package sessions

import (
	"context"
	redisComponent "f1-game/internal/component/redis"
	nameActor "f1-game/internal/name/actor"
	nameRemote "f1-game/internal/name/remote"
	"f1-game/internal/pb"
	"fmt"
	"sync/atomic"
	"time"

	cstring "github.com/cherry-game/cherry/extend/string"
	clog "github.com/cherry-game/cherry/logger"
	"github.com/cherry-game/cherry/net/parser/pomelo"
	cproto "github.com/cherry-game/cherry/net/proto"
	cprofile "github.com/cherry-game/cherry/profile"
	"github.com/go-redis/redis/v8"
	"github.com/goburrow/cache"
)

var (
	actor = &Actor{}
)

type (
	Actor struct {
		pomelo.ActorBase
		isSync           bool            // 是否实时同步标识
		cache            cache.Cache     // session cache.  key:playeID, value:*cproto.Session
		count            int64           // session count
		redisClient      *redis.Client   // redis client
		ctx              context.Context // context
		redisFields      []string        // redis查询的字段
		redisKeyFormat   string          // redis key
		onbindFuncList   []OnBindFunc    // 绑定触发函数
		onUnbindFuncList []OnUnbindFunc  // 解绑触发函数
	}

	OnBindFunc   func(session *cproto.Session) bool
	OnUnbindFunc func(playerID int64)
)

// isSync = true 则实时同步session到本地进程
func NewActor(isSync bool) *Actor {
	actor.isSync = isSync
	actor.ctx = context.Background()
	actor.redisFields = []string{
		UID,
		SID,
		AgentPath,
	}
	actor.redisKeyFormat = fmt.Sprintf("%s-game:session:", cprofile.Env())
	actor.cache = buildCache(isSync, 100000, 1*time.Second)
	actor.count = 0

	return actor
}

func (p *Actor) AliasID() string {
	return nameActor.GateSessions
}

func (p *Actor) OnInit() {
	if p.isSync {
		p.Remote().Register(nameRemote.Sessions_Bind, p.bind)
		p.Remote().Register(nameRemote.Sessions_Unbind, p.ubind)
	}

	// 初始化redis
	p.initRedis()
	clog.Infof("[sessions] actor init success. isSync = %v", p.isSync)
}

func (p *Actor) bind(session *cproto.Session) {
	playerID := GetPlayerID(session)
	if playerID > 0 {
		p.addSession2Cache(playerID, session)

		for _, fn := range p.onbindFuncList {
			fn(session)
		}
	}
}

func (p *Actor) ubind(req *pb.I64) {
	if req.Value > 0 {
		p.delSessoin2Cache(req.Value)

		for _, fn := range p.onUnbindFuncList {
			fn(req.Value)
		}
	}
}

func (p *Actor) initRedis() {
	config := cprofile.GetConfig("sessions")
	if config.LastError() != nil {
		clog.Fatal("sessions property not exists.")
		return
	}

	redisID := config.GetString("redis_id")
	if redisID == "" {
		clog.Fatal("sessions redis_id property not exists.")
		return
	}

	redisConfigList := cprofile.GetConfig("redis")
	if redisConfigList.LastError() != nil {
		clog.Fatal("redis property not exists.")
		return
	}

	redisConfig := redisComponent.GetConfigWithRedisID(redisConfigList, redisID)
	if redisConfig == nil {
		clog.Fatalf("redis config not found. [redisID = %s]", redisID)
		return
	}

	_, options := redisComponent.NewClientOption(redisConfig)

	p.redisClient = redis.NewClient(options)
}

func (p *Actor) BuildPlayerKey(playerID int64) string {
	return fmt.Sprintf("%s%d", p.redisKeyFormat, playerID)
}

func (p *Actor) GetCacheSession(playerID int64) (*cproto.Session, bool) {
	value, found := p.cache.GetIfPresent(playerID)
	if !found {
		return nil, false
	}

	return value.(*cproto.Session), true
}

func (p *Actor) GetCacheSessions(playerIDS []int64) (map[int64]*cproto.Session, []int64) {
	var (
		sessionMaps   = map[int64]*cproto.Session{}
		missPlayerIDS []int64
	)

	for _, playerID := range playerIDS {
		if session, found := p.GetCacheSession(playerID); found {
			sessionMaps[playerID] = session
		} else {
			if playerID > 0 {
				missPlayerIDS = append(missPlayerIDS, playerID)
			}
		}
	}

	return sessionMaps, missPlayerIDS
}

func (p *Actor) GetRedisSession(playerID int64) (*cproto.Session, bool) {
	key := p.BuildPlayerKey(playerID)

	result, err := p.redisClient.HMGet(p.ctx, key, p.redisFields...).Result()
	if err != nil {
		clog.Errorf("[GetRedisSession] error: %v", err)
		return nil, false
	}

	session := result2Session(result)
	if len(session.AgentPath) > 0 {
		p.addSession2Cache(playerID, session) // add session to cache
		return session, true
	}

	return nil, false
}

func (p *Actor) GetRedisSessions(playerIDS []int64) map[int64]*cproto.Session {
	pipeline := p.redisClient.Pipeline()
	defer pipeline.Close()

	cmds := make([]*redis.SliceCmd, 0, len(playerIDS))
	for _, playerID := range playerIDS {
		key := p.BuildPlayerKey(playerID)
		cmds = append(cmds, pipeline.HMGet(p.ctx, key, p.redisFields...))
	}

	if _, err := pipeline.Exec(p.ctx); err != nil {
		clog.Errorf("[GetRedisSessions] error: %v", err)
		return nil
	}

	sessionMap := map[int64]*cproto.Session{}

	for i, playerID := range playerIDS {
		result, err := cmds[i].Result()
		if err != nil {
			continue
		}

		session := result2Session(result)
		if len(session.AgentPath) > 0 && session.Uid > 0 {
			p.addSession2Cache(playerID, session) // add session to cache
			sessionMap[playerID] = session
		}
	}

	return sessionMap
}

// 设置session绑定时触发的函数
func SetOnBindFunc(fn OnBindFunc) {
	if fn != nil {
		actor.onbindFuncList = append(actor.onbindFuncList, fn)
	}
}

// 设置session解绑时触发的函数
func SetOnUnbindFunc(fn OnUnbindFunc) {
	if fn != nil {
		actor.onUnbindFuncList = append(actor.onUnbindFuncList, fn)
	}
}

func buildCache(isSync bool, maxSize int, expire time.Duration) cache.Cache {
	var options []cache.Option

	if !isSync {
		// 非实时同步模式，缓存后x秒过期
		options = append(options, cache.WithExpireAfterWrite(expire))
	}

	// 缓存最多10万个session
	options = append(options, cache.WithMaximumSize(maxSize))

	// 缓存失效时触发监听函数
	options = append(options, cache.WithRemovalListener(func(key cache.Key, _ cache.Value) {
		// playerID := key.(int64)
		// session数量-1
		atomic.AddInt64(&actor.count, -1)
	}))

	return cache.New(options...)
}

func result2Session(result []interface{}) *cproto.Session {
	uid := cstring.ToInt64D(cstring.ToString(result[0]), 0)
	sid := cstring.ToString(result[1])
	agentPath := cstring.ToString(result[2])

	return &cproto.Session{
		Uid:       uid,
		Sid:       sid,
		AgentPath: agentPath,
	}
}

func (p *Actor) addSession2Cache(playerID int64, session *cproto.Session) {
	p.cache.Put(playerID, session)
	atomic.AddInt64(&p.count, 1)
}

func (p *Actor) delSessoin2Cache(playerID int64) {
	p.cache.Invalidate(playerID)
}