xpggame/call/session.go

package call

import (
	"context"
	"fmt"
	"server/common"
	"server/db"
	"server/natsClient"
	"server/pb"
	"server/util"
	"strconv"

	"github.com/gogo/protobuf/proto"
	"github.com/liangdas/mqant/gate"
	basegate "github.com/liangdas/mqant/gate/base"
	"github.com/liangdas/mqant/log"
	"github.com/liangdas/mqant/module"
	mqrpc "github.com/liangdas/mqant/rpc"
	"github.com/liangdas/mqant/selector"
)

var (
	caller module.Module
)

// NewCaller 新建一个调用对象
func NewCaller(m module.Module) {
	caller = m
}

func GetCaller() module.Module {
	return caller
}

func GetTopicName() string {
	name := caller.GetType()
	return name
}

// SendSS 给玩家发送消息
func SendSS(session gate.Session, pid int, data proto.Message, t ...string) error {
	payload := []byte{}
	var err error
	if data != nil {
		payload, err = proto.Marshal(data)
		if err != nil {
			log.Error("err:%v", err)
			return err
		}
	}
	moduleName := ""
	if t != nil {
		moduleName = t[0]
	} else {
		moduleName = GetTopicName()
	}
	// moduleType := pb.ServerType_name[pb.ModuleType_value[moduleName]]
	// moduleType := GetModuleID(moduleName)
	topic := fmt.Sprintf("%v:%v", moduleName, strconv.Itoa(pid))
	// log.Debug("topic:%v", topic)
	session.SendNR(topic, payload)
	return nil
}

// SendSSBytes 给玩家发送消息
func SendSSBytes(session gate.Session, pid int, payload []byte) error {
	// moduleType := GetModuleID(caller.GetType())
	// moduleType := pb.ServerType_name[pb.ModuleType_value[caller.GetType()]]
	topic := fmt.Sprintf("%v:%v", GetTopicName(), pid)
	session.SendNR(topic, payload)
	return nil
}

// SendNR 给玩家发送消息 注意：必须先调用NewCaller初始化
func SendNR(uid int, pid int, data proto.Message, t ...string) {
	util.Go(func() {
		payload, err := proto.Marshal(data)
		if err != nil {
			log.Error("err:%v", err)
			return
		}
		moduleName := ""
		if t != nil {
			moduleName = t[0]
		} else {
			moduleName = GetTopicName()
		}
		// moduleType := GetModuleID(moduleName)
		// moduleType := pb.ServerType_name[pb.ModuleType_value[moduleName]]
		topic := fmt.Sprintf("%v:%v", moduleName, pid)

		g := GetUserSession(uid)
		if g == nil {
			// log.Error("get session fail")
			return
		}
		if err := g.Send(topic, payload); err != "" {
			log.Error("err:%v", err)
			return
		}
	})
}

// Publish 发布消息
func Publish(topic string, data proto.Message) error {
	send, _ := proto.Marshal(data)
	err := caller.GetApp().Transport().Publish(topic, send)
	if err != nil {
		log.Error("err:%v", err)
		return err
	}
	return nil
}

// Publish 发布消息
func PublishRequest(topic, reply string, data proto.Message) error {
	send, _ := proto.Marshal(data)
	return caller.GetApp().Transport().PublishRequest(topic, reply, send)
}

// RPC 服务器内部调用
func RPC(module, method string, param proto.Message, opts ...selector.SelectOption) (interface{}, string) {
	return caller.GetApp().Call(context.Background(), module, method, mqrpc.Param(param), opts...)
}

func GetUserSession(uid int) gate.Session {
	s := db.Redis().GetUserSession(uid)
	if s == nil {
		// log.Error("get session fail")
		return nil
	}
	g, _ := basegate.NewSession(caller.GetApp(), nil)
	g.SetSessionID(s.SessionID)
	g.SetServerID(s.GateID)
	g.SetUserID(strconv.Itoa(uid))
	return g
}

// Broadcast 发送广播
func BroadcastReq(one *common.ConfigBroadcast, content ...string) {
	c := one.Content
	for _, v := range content {
		c = v
	}
	data := &pb.InnerBroadcast{
		ID:        int32(one.ID),
		Content:   c,
		Priority:  int32(one.Priority),
		Frequency: int32(one.LoopFrequency),
		Interval:  int32(one.Interval),
	}
	// log.Debug("broadcastReq:%+v", data)
	send, _ := proto.Marshal(data)
	caller.GetApp().Transport().Publish(natsClient.TopicBroadcastReq, send)
}