上传聊天模块以及底层消息广播接口

lego/base/core.go

@@ -77,6 +77,7 @@ type IRPCXService interface {
 	RegisterFunction(fn interface{}) (err error)
 	RegisterFunctionName(name string, fn interface{}) (err error)
 	RpcCall(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error)
+	RpcBroadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error)
 	RpcGo(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (call *client.Call, err error)
 	AcrossClusterRpcCall(ctx context.Context, clusterTag string, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error)
 	AcrossClusterRpcGo(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (call *client.Call, err error)

lego/base/rpcx/service.go

@@ -148,6 +148,12 @@ func (this *RPCXService) RpcCall(ctx context.Context, servicePath string, servic
 	return rpcx.Call(ctx, servicePath, serviceMethod, args, reply)
 }
 
+///广播 执行目标远程服务方法
+///servicePath = worker  表示采用负载的方式调用 worker类型服务执行rpc方法
+func (this *RPCXService) RpcBroadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error) {
+	return rpcx.Broadcast(ctx, servicePath, serviceMethod, args, reply)
+}
+
 ///异步 执行目标远程服务方法
 ///servicePath = worker  表示采用负载的方式调用 worker类型服务执行rpc方法
 ///servicePath = worker/worker_1  表示寻找目标服务节点调用rpc方法

lego/core/core.go

@@ -44,6 +44,7 @@ type ServiceSttings struct {
 
 //基础服务的接口设计
 type IService interface {
+	GetTag() string                                             //获取集群id
 	GetId() string                                              //获取服务id
 	GetType() string                                            //获取服务类型
 	GetVersion() string                                         //获取服务版本

lego/sys/rpcx/client.go

@@ -175,6 +175,42 @@ func (this *Client) Go(ctx context.Context, servicePath string, serviceMethod st
 	return c.Go(ctx, string(serviceMethod), args, reply, done)
 }
 
+//异步调用
+func (this *Client) Broadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error) {
+	if servicePath == "" {
+		err = errors.New("servicePath no cant null")
+		return
+	}
+	var (
+		spath []string
+		d     *client.ConsulDiscovery
+		c     client.XClient
+		ok    bool
+	)
+	spath = strings.Split(servicePath, "/")
+	if c, ok = this.clients[spath[0]]; !ok {
+		if d, err = client.NewConsulDiscovery(this.options.ServiceTag, spath[0], this.options.ConsulServers, nil); err != nil {
+			return
+		}
+		c = client.NewBidirectionalXClient(spath[0], client.Failfast, client.RandomSelect, d, client.DefaultOption, this.msgChan)
+		c.GetPlugins().Add(this)
+		if this.options.RpcxStartType == RpcxStartByClient && this.options.AutoConnect {
+			c.SetSelector(newSelector(this.UpdateServer))
+		} else {
+			c.SetSelector(newSelector(nil))
+		}
+		this.clients[spath[0]] = c
+	}
+	ctx = context.WithValue(ctx, share.ReqMetaDataKey, map[string]string{
+		ServiceClusterTag: this.options.ServiceTag,
+		CallRoutRulesKey:  servicePath,
+		ServiceAddrKey:    "tcp@" + this.options.ServiceAddr,
+		ServiceMetaKey:    this.metadata,
+	})
+	err = c.Broadcast(ctx, serviceMethod, args, reply)
+	return
+}
+
 //跨集群 同步调用
 func (this *Client) AcrossClusterCall(ctx context.Context, clusterTag string, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error) {
 	if servicePath == "" {

lego/sys/rpcx/core.go

@@ -26,10 +26,15 @@ type (
 		RegisterFunctionName(name string, fn interface{}) (err error)
 		UnregisterAll() (err error)
 		Call(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error)
+		Broadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error)
 		Go(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}, done chan *client.Call) (call *client.Call, err error)
 		AcrossClusterCall(ctx context.Context, clusterTag string, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error)
 		AcrossClusterGo(ctx context.Context, clusterTag string, servicePath string, serviceMethod string, args interface{}, reply interface{}, done chan *client.Call) (call *client.Call, err error)
 	}
+	ISelector interface {
+		client.Selector
+		Find(ctx context.Context, servicePath, serviceMethod string, args interface{}) []string
+	}
 )
 
 var (
@@ -77,6 +82,10 @@ func Call(ctx context.Context, servicePath string, serviceMethod string, args in
 	return defsys.Call(ctx, servicePath, serviceMethod, args, reply)
 }
 
+func Broadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error) {
+	return defsys.Broadcast(ctx, servicePath, serviceMethod, args, reply)
+}
+
 func Go(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}, done chan *client.Call) (call *client.Call, err error) {
 	return defsys.Go(ctx, servicePath, serviceMethod, args, reply, done)
 }

lego/sys/rpcx/rpcx.go

@@ -65,6 +65,11 @@ func (this *RPCX) Call(ctx context.Context, servicePath string, serviceMethod st
 	return this.client.Call(ctx, servicePath, serviceMethod, args, reply)
 }
 
+//广播调用
+func (this *RPCX) Broadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error) {
+	return this.client.Broadcast(ctx, servicePath, serviceMethod, args, reply)
+}
+
 //异步调用
 func (this *RPCX) Go(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}, done chan *client.Call) (call *client.Call, err error) {
 	return this.client.Go(ctx, servicePath, serviceMethod, args, reply, done)

lego/sys/rpcx/selector.go

@@ -68,6 +68,17 @@ func (this *Selector) Select(ctx context.Context, servicePath, serviceMethod str
 	return ""
 }
 
+//找到同类型节点信息
+func (this *Selector) Find(ctx context.Context, servicePath, serviceMethod string, args interface{}) []string {
+	if nodes, ok := this.serversType[servicePath]; ok {
+		addrs := make([]string, len(nodes))
+		for i, v := range nodes {
+			addrs[i] = v.ServiceAddr
+		}
+	}
+	return nil
+}
+
 //更新服务列表
 func (this *Selector) UpdateServer(servers map[string]string) {
 	ss := make(map[string]*ServiceNode)

lego/sys/rpcx/service.go

@@ -23,7 +23,7 @@ func newService(options *Options) (sys *Service, err error) {
 		options:    options,
 		metadata:   fmt.Sprintf("stag=%s&stype=%s&sid=%s&version=%s&addr=%s", options.ServiceTag, options.ServiceType, options.ServiceId, options.ServiceVersion, "tcp@"+options.ServiceAddr),
 		server:     server.NewServer(),
-		selectors:  make(map[string]client.Selector),
+		selectors:  make(map[string]ISelector),
 		clients:    make(map[string]net.Conn),
 		clientmeta: make(map[string]string),
 		pending:    make(map[uint64]*client.Call),
@@ -49,8 +49,8 @@ type Service struct {
 	options     *Options
 	metadata    string
 	server      *server.Server
-	selectors   map[string]client.Selector
+	selectors   map[string]ISelector
-	clientmutex sync.Mutex
+	clientmutex sync.RWMutex
 	clients     map[string]net.Conn
 	clientmeta  map[string]string
 	mutex       sync.Mutex // protects following
@@ -128,6 +128,12 @@ func (this *Service) Call(ctx context.Context, servicePath string, serviceMethod
 	return
 }
 
+//广播调用
+func (this *Service) Broadcast(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error) {
+	err = this.broadcast(ctx, this.options.ServiceTag, servicePath, serviceMethod, args, reply)
+	return
+}
+
 //异步调用 远程服务
 func (this *Service) Go(ctx context.Context, servicePath string, serviceMethod string, args interface{}, reply interface{}, done chan *client.Call) (_call *client.Call, err error) {
 	_, _call, err = this.call(ctx, this.options.ServiceTag, servicePath, serviceMethod, args, reply, done)
@@ -327,10 +333,13 @@ func (this *Service) call(ctx context.Context, clusterTag string, servicePath st
 	if selector, ok = this.selectors[clusterTag]; !ok {
 		err = fmt.Errorf("on found serviceTag:%s", clusterTag)
 	}
+	this.clientmutex.RLock()
 	if clientaddr = selector.Select(ctx, spath[0], serviceMethod, args); clientaddr == "" {
 		err = fmt.Errorf("on found servicePath:%s", servicePath)
+		this.clientmutex.RUnlock()
 		return
 	}
+	this.clientmutex.RUnlock()
 	if conn, ok = this.clients[clientaddr]; !ok {
 		err = fmt.Errorf("on found clientaddr:%s", clientaddr)
 		return
@@ -353,6 +362,104 @@ func (this *Service) call(ctx context.Context, clusterTag string, servicePath st
 	return
 }
 
+//执行远程调用
+func (this *Service) broadcast(ctx context.Context, clusterTag string, servicePath string, serviceMethod string, args interface{}, reply interface{}) (err error) {
+	var (
+		spath       []string
+		clientaddrs []string
+		metadata    map[string]string
+		selector    ISelector
+		ok          bool
+	)
+	if servicePath == "" {
+		err = errors.New("servicePath no cant null")
+		return
+	}
+	metadata = map[string]string{
+		ServiceClusterTag: clusterTag,
+		CallRoutRulesKey:  servicePath,
+		ServiceAddrKey:    "tcp@" + this.options.ServiceAddr,
+		ServiceMetaKey:    this.metadata,
+	}
+	spath = strings.Split(servicePath, "/")
+	ctx = context.WithValue(ctx, share.ReqMetaDataKey, map[string]string{
+		CallRoutRulesKey: servicePath,
+		ServiceAddrKey:   "tcp@" + this.options.ServiceAddr,
+		ServiceMetaKey:   this.metadata,
+	})
+	if selector, ok = this.selectors[clusterTag]; !ok {
+		err = fmt.Errorf("on found serviceTag:%s", clusterTag)
+	}
+	if clientaddrs = selector.Find(ctx, spath[0], serviceMethod, args); clientaddrs == nil || len(clientaddrs) == 0 {
+		err = fmt.Errorf("on found servicePath:%s", servicePath)
+		return
+	}
+	l := len(clientaddrs)
+	done := make(chan error, l)
+	for _, v := range clientaddrs {
+		go func(addr string) {
+
+			this.clientmutex.RLock()
+			conn, ok := this.clients[addr]
+			if !ok {
+				err = fmt.Errorf("on found clientaddr:%s", addr)
+				this.clientmutex.RUnlock()
+				return
+			}
+			this.clientmutex.RUnlock()
+			_call := new(client.Call)
+			_call.ServicePath = servicePath
+			_call.ServiceMethod = serviceMethod
+			_call.Args = args
+			_call.Reply = reply
+			_call.Done = make(chan *client.Call, 10)
+			this.send(ctx, conn, spath[0], serviceMethod, metadata, _call)
+			seq, _ := ctx.Value(seqKey{}).(*uint64)
+			select {
+			case <-ctx.Done(): // cancel by context
+				this.mutex.Lock()
+				call := this.pending[*seq]
+				delete(this.pending, *seq)
+				this.mutex.Unlock()
+				if call != nil {
+					call.Error = ctx.Err()
+					call.Done <- call
+				}
+				done <- ctx.Err()
+			case call := <-_call.Done:
+				err = call.Error
+				meta := ctx.Value(share.ResMetaDataKey)
+				if meta != nil && len(call.ResMetadata) > 0 {
+					resMeta := meta.(map[string]string)
+					for k, v := range call.ResMetadata {
+						resMeta[k] = v
+					}
+					resMeta[share.ServerAddress] = conn.RemoteAddr().String()
+				}
+				done <- nil
+			}
+		}(v)
+	}
+
+	timeout := time.NewTimer(time.Minute)
+check:
+	for {
+		select {
+		case err = <-done:
+			l--
+			if l == 0 || err != nil { // all returns or some one returns an error
+				break check
+			}
+		case <-timeout.C:
+			err = errors.New(("timeout"))
+			break check
+		}
+	}
+
+	timeout.Stop()
+	return err
+}
+
 //发送远程调用请求
 func (this *Service) send(ctx context.Context, conn net.Conn, servicePath string, serviceMethod string, metadata map[string]string, call *client.Call) {
 	defer func() {

modules/chat/api.go

@@ -0,0 +1,29 @@
+package chat
+
+import (
+	"go_dreamfactory/modules"
+
+	"go_dreamfactory/lego/core"
+)
+
+/*
+API
+*/
+type apiComp struct {
+	modules.MCompGate
+	service core.IService
+	module  *Chat
+}
+
+//组件初始化接口
+func (this *apiComp) Init(service core.IService, module core.IModule, comp core.IModuleComp, options core.IModuleOptions) (err error) {
+	this.MCompGate.Init(service, module, comp, options)
+	this.module = module.(*Chat)
+	this.service = service
+	return
+}
+
+func (this *apiComp) Start() (err error) {
+	err = this.MCompGate.Start()
+	return
+}

modules/chat/api_getlist.go

@@ -0,0 +1,28 @@
+package chat
+
+import (
+	"go_dreamfactory/comm"
+	"go_dreamfactory/pb"
+
+	"google.golang.org/protobuf/proto"
+)
+
+//参数校验
+func (this *apiComp) GetListCheck(session comm.IUserSession, req *pb.ChatGetListReq) (code pb.ErrorCode) {
+
+	return
+}
+
+///获取未读消息
+func (this *apiComp) GetList(session comm.IUserSession, req *pb.ChatGetListReq) (code pb.ErrorCode, data proto.Message) {
+	var (
+		err  error
+		list []*pb.DBChat
+	)
+	if list, err = this.module.modelChat.QueryUserMsg(session.GetUserId()); err != nil {
+		code = pb.ErrorCode_DBError
+		return
+	}
+	session.SendMsg(string(this.module.GetType()), "getlist", &pb.ChatGetListResp{Chats: list})
+	return
+}

modules/chat/api_send.go

@@ -0,0 +1,58 @@
+package chat
+
+import (
+	"go_dreamfactory/comm"
+	"go_dreamfactory/pb"
+	"time"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"google.golang.org/protobuf/proto"
+)
+
+//参数校验
+func (this *apiComp) SendCheck(session comm.IUserSession, req *pb.ChatSendReq) (code pb.ErrorCode) {
+
+	return
+}
+
+///消息发送请求
+func (this *apiComp) Send(session comm.IUserSession, req *pb.ChatSendReq) (code pb.ErrorCode, data proto.Message) {
+	var (
+		err  error
+		msg  *pb.DBChat
+		user *pb.DBUser
+	)
+	msg = &pb.DBChat{
+		Id:      primitive.NewObjectID().Hex(),
+		Channel: req.Channel,
+		Suid:    session.GetUserId(),
+		Ruid:    req.TargetId,
+		AreaId:  this.service.GetTag(),
+		Headid:  user.Avatar,
+		Content: req.Content,
+		Ctime:   time.Now().Unix(),
+	}
+	if err = this.module.modelChat.AddChatMsg(msg); err != nil {
+		code = pb.ErrorCode_DBError
+		return
+	}
+	session.SendMsg(string(this.module.GetType()), "send", &pb.ChatSendResp{})
+	switch msg.Channel {
+	case pb.ChatChannel_World:
+		this.module.PushWorld(msg)
+		break
+	case pb.ChatChannel_Union:
+		this.module.PushUnion(msg)
+		break
+	case pb.ChatChannel_Private:
+		this.module.PushUser(msg)
+		break
+	case pb.ChatChannel_CrossServer:
+		this.module.PushAllWorld(msg)
+		break
+	case pb.ChatChannel_System:
+		this.module.PushAllWorld(msg)
+		break
+	}
+	return
+}

modules/chat/api_spansend.go

@@ -0,0 +1,26 @@
+package chat
+
+import (
+	"go_dreamfactory/comm"
+	"go_dreamfactory/pb"
+
+	"google.golang.org/protobuf/proto"
+)
+
+//参数校验
+func (this *apiComp) SpanSendCheck(session comm.IUserSession, req *pb.ChatSpanSendReq) (code pb.ErrorCode) {
+
+	return
+}
+
+///跨越服务消息请求
+func (this *apiComp) SpanSend(session comm.IUserSession, req *pb.ChatSpanSendReq) (code pb.ErrorCode, data proto.Message) {
+
+	defer func() {
+		if code == pb.ErrorCode_Success {
+			session.SendMsg(string(this.module.GetType()), "send", &pb.ChatSendResp{})
+		}
+	}()
+
+	return
+}

modules/chat/configure.go

@@ -0,0 +1,23 @@
+package chat
+
+import (
+	"go_dreamfactory/modules"
+
+	"go_dreamfactory/lego/core"
+)
+
+const (
+	game_equipment = "game_equipment.json"
+)
+
+///背包配置管理组件
+type configureComp struct {
+	modules.MCompConfigure
+}
+
+//组件初始化接口
+func (this *configureComp) Init(service core.IService, module core.IModule, comp core.IModuleComp, options core.IModuleOptions) (err error) {
+	this.ModuleCompBase.Init(service, module, comp, options)
+
+	return
+}

modules/chat/modelChat.go

@@ -0,0 +1,60 @@
+package chat
+
+import (
+	"context"
+	"go_dreamfactory/lego/core"
+	"go_dreamfactory/modules"
+	"go_dreamfactory/pb"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/x/bsonx"
+)
+
+///论坛 数据组件
+type modelChatComp struct {
+	modules.MCompModel
+	module *Chat
+}
+
+//组件初始化接口
+func (this *modelChatComp) Init(service core.IService, module core.IModule, comp core.IModuleComp, opt core.IModuleOptions) (err error) {
+	this.MCompModel.Init(service, module, comp, opt)
+	this.module = module.(*Chat)
+	this.TableName = "chat"
+	//创建uid索引
+	this.DB.CreateIndex(core.SqlTable(this.TableName), mongo.IndexModel{
+		Keys: bsonx.Doc{{Key: "ruid", Value: bsonx.Int32(1)}},
+	})
+	return
+}
+
+//查询用户未读消息
+func (this *modelChatComp) QueryUserMsg(uid string) (result []*pb.DBChat, err error) {
+	var (
+		c *mongo.Cursor
+	)
+	if c, err = this.DB.Find(core.SqlTable(this.TableName), bson.M{"ruid": uid}); err != nil {
+		this.module.Errorf("err:%v", err)
+		return
+	}
+	result = make([]*pb.DBChat, c.RemainingBatchLength())
+	n := 0
+	for c.Next(context.TODO()) {
+		temp := &pb.DBChat{}
+		if err = c.Decode(temp); err != nil {
+			this.module.Errorf("err:%v", err)
+			return
+		}
+		result[n] = temp
+		n++
+	}
+	this.DB.DeleteMany(core.SqlTable(this.TableName), bson.M{"ruid": uid}) //清理数据
+	return
+}
+
+//添加聊天消息到数据库中
+func (this *modelChatComp) AddChatMsg(msg *pb.DBChat) (err error) {
+
+	return
+}

modules/chat/module.go

@@ -0,0 +1,89 @@
+package chat
+
+import (
+	"context"
+	"go_dreamfactory/comm"
+	"go_dreamfactory/lego/base"
+	"go_dreamfactory/lego/core"
+	"go_dreamfactory/modules"
+	"go_dreamfactory/pb"
+
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+/*
+模块名:论坛
+描述:处理跨服社交论坛相关业务
+开发:李伟
+*/
+func NewModule() core.IModule {
+	m := new(Chat)
+	return m
+}
+
+type Chat struct {
+	modules.ModuleBase
+	service   base.IRPCXService //rpc服务对象 通过这个对象可以发布服务和调用其他服务的接口
+	api_comp  *apiComp
+	configure *configureComp
+	modelChat *modelChatComp
+}
+
+//模块名
+func (this *Chat) GetType() core.M_Modules {
+	return comm.ModuleEquipment
+}
+
+//模块初始化接口 注册用户创建角色事件
+func (this *Chat) Init(service core.IService, module core.IModule, options core.IModuleOptions) (err error) {
+	err = this.ModuleBase.Init(service, module, options)
+	this.service = service.(base.IRPCXService)
+	return
+}
+
+//装备组件
+func (this *Chat) OnInstallComp() {
+	this.ModuleBase.OnInstallComp()
+	this.api_comp = this.RegisterComp(new(apiComp)).(*apiComp)
+	this.modelChat = this.RegisterComp(new(modelChatComp)).(*modelChatComp)
+	this.configure = this.RegisterComp(new(configureComp)).(*configureComp)
+}
+
+//Push--------------------------------------------------------------------------------------------------------------
+//推送消息到世界
+func (this *Chat) PushWorld(msg *pb.DBChat) {
+	var (
+		err   error
+		reply *pb.RPCMessageReply
+	)
+	reply = &pb.RPCMessageReply{}
+	data, _ := anypb.New(msg)
+	if err = this.service.RpcBroadcast(context.Background(), comm.Service_Gateway, string(comm.Rpc_GatewaySendRadioMsg), pb.UserMessage{
+		MainType: string(this.GetType()),
+		SubType:  "push",
+		Data:     data,
+	}, reply); err != nil {
+		this.Errorf("err:%v", err)
+	}
+}
+
+//推送消息到工会
+func (this *Chat) PushUnion(msg *pb.DBChat) {
+
+}
+
+//推送消息到用户
+func (this *Chat) PushUser(msg *pb.DBChat) {
+	if session, ok := this.GetUserSession(msg.Ruid); ok {
+		session.SendMsg(string(this.GetType()), "push", msg)
+		if err := session.Push(); err != nil {
+			this.Errorf("err:%v", err)
+		}
+		return
+	}
+}
+
+//推送消息到所有区服
+func (this *Chat) PushAllWorld(msg *pb.DBChat) {
+
+}

modules/chat/module_test.go

@@ -0,0 +1,75 @@
+package chat_test
+
+import (
+	"fmt"
+	"go_dreamfactory/comm"
+	"go_dreamfactory/lego"
+	"go_dreamfactory/lego/base/rpcx"
+	"go_dreamfactory/lego/core"
+	"go_dreamfactory/lego/sys/log"
+	"go_dreamfactory/modules/chat"
+	"go_dreamfactory/services"
+	"go_dreamfactory/sys/cache"
+	"go_dreamfactory/sys/configure"
+	"go_dreamfactory/sys/db"
+	"os"
+	"testing"
+	"time"
+)
+
+func newService(ops ...rpcx.Option) core.IService {
+	s := new(TestService)
+	s.Configure(ops...)
+	return s
+}
+
+//梦工厂基础服务对象
+type TestService struct {
+	rpcx.RPCXService
+}
+
+//初始化相关系统
+func (this *TestService) InitSys() {
+	this.RPCXService.InitSys()
+	if err := cache.OnInit(this.GetSettings().Sys["cache"]); err != nil {
+		panic(fmt.Sprintf("init sys.cache err: %s", err.Error()))
+	} else {
+		log.Infof("init sys.cache success!")
+	}
+	if err := db.OnInit(this.GetSettings().Sys["db"]); err != nil {
+		panic(fmt.Sprintf("init sys.db err: %s", err.Error()))
+	} else {
+		log.Infof("init sys.db success!")
+	}
+	if err := configure.OnInit(this.GetSettings().Sys["configure"]); err != nil {
+		panic(fmt.Sprintf("init sys.configure err: %s", err.Error()))
+	} else {
+		log.Infof("init sys.configure success!")
+	}
+}
+
+var service core.IService
+var s_gateComp comm.ISC_GateRouteComp = services.NewGateRouteComp()
+var module = new(chat.Chat)
+
+//测试环境下初始化db和cache 系统
+func TestMain(m *testing.M) {
+	service = newService(
+		rpcx.SetConfPath("../../bin/conf/worker_1.yaml"),
+		rpcx.SetVersion("1.0.0.0"),
+	)
+	service.OnInstallComp( //装备组件
+		s_gateComp, //此服务需要接受用户的消息 需要装备网关组件
+	)
+	go func() {
+		lego.Run(service, //运行模块
+			module,
+		)
+	}()
+	time.Sleep(time.Second * 3)
+	defer os.Exit(m.Run())
+}
+
+func Test_Module(t *testing.T) {
+
+}

modules/comp_model.go

@@ -270,97 +270,6 @@ func (this *MCompModel) Get(uid string, data interface{}) (err error) {
 	return
 }
 
-//获取列表数据 注意 data 必须是 切片的指针 *[]type 暂时废弃
-func (this *MCompModel) mGetList(uid string, data interface{}) (err error) {
-	var keys map[string]string = make(map[string]string)
-	var c *mongo.Cursor
-	t := reflect.TypeOf(data)
-	if t.Kind() == reflect.Ptr {
-		t = t.Elem()
-	}
-
-	if t.Kind() == reflect.Slice {
-		t = t.Elem()
-	} else {
-		err = fmt.Errorf("Input param is not a slice")
-	}
-	sl := reflect.ValueOf(data)
-	if t.Kind() == reflect.Ptr {
-		sl = sl.Elem()
-	}
-	st := sl.Type()
-	sliceType := st.Elem()
-	if sliceType.Kind() == reflect.Ptr {
-		sliceType = sliceType.Elem()
-	}
-	err = this.Redis.HGetAll(this.ukey(uid), keys)
-	if err == nil {
-		for _, v := range keys {
-			if sl.Len() < sl.Cap() {
-				sl.Set(sl.Slice(0, sl.Len()+1))
-				elem := sl.Index(sl.Len() - 1)
-				if elem.IsNil() {
-					elem.Set(reflect.New(sliceType))
-				}
-				if err = this.Redis.HGetAll(v, elem.Elem().Addr().Interface()); err != nil {
-					return
-				}
-				continue
-			}
-			elem := reflect.New(sliceType)
-			sl.Set(reflect.Append(sl, elem))
-			if err = this.Redis.HGetAll(v, elem.Elem().Addr().Interface()); err != nil {
-				return
-			}
-		}
-	}
-	if err == redis.RedisNil {
-		//query from mgo
-		if c, err = this.DB.Find(core.SqlTable(this.TableName), bson.M{"uid": uid}); err != nil {
-			return err
-		} else {
-			var temp map[string]interface{} = make(map[string]interface{})
-			var keys map[string]string = make(map[string]string)
-			for c.Next(context.Background()) {
-				_id := c.Current.Lookup("_id").StringValue()
-				if sl.Len() < sl.Cap() {
-					sl.Set(sl.Slice(0, sl.Len()+1))
-					elem := sl.Index(sl.Len() - 1)
-					if elem.IsNil() {
-						elem.Set(reflect.New(sliceType))
-					}
-					if err = c.Decode(elem.Elem().Addr().Interface()); err != nil {
-						return
-					}
-					temp[_id] = elem.Elem().Addr().Interface()
-					continue
-				}
-				elem := reflect.New(sliceType)
-				sl.Set(reflect.Append(sl, elem))
-				if err = c.Decode(elem.Elem().Addr().Interface()); err != nil {
-					return
-				}
-				temp[_id] = elem.Elem().Addr().Interface()
-			}
-			if len(temp) == 0 { //没有数据自己返回
-				return
-			}
-			for k, v := range temp {
-				key := this.ukeylist(uid, k)
-				if err = this.Redis.HMSet(key, v); err != nil {
-					return
-				}
-				keys[k] = key
-			}
-			if err = this.Redis.HMSet(this.ukey(uid), keys); err != nil {
-				return
-			}
-		}
-	}
-
-	return err
-}
-
 //获取列表数据 注意 data 必须是 切片的指针 *[]type
 func (this *MCompModel) GetList(uid string, data interface{}) (err error) {
 	var (

modules/equipment/api_equip.go

@@ -28,11 +28,6 @@ func (this *apiComp) Equip(session comm.IUserSession, req *pb.EquipmentEquipReq)
 		hero           *pb.DBHero
 	)
 
-	defer func() {
-		if code == pb.ErrorCode_Success {
-			session.SendMsg(string(this.module.GetType()), "equip", &pb.EquipmentEquipResp{Equipments: equipments})
-		}
-	}()
 	if code = this.EquipCheck(session, req); code != pb.ErrorCode_Success {
 		return
 	}
@@ -124,5 +119,6 @@ func (this *apiComp) Equip(session comm.IUserSession, req *pb.EquipmentEquipReq)
 			return
 		}
 	}
+	session.SendMsg(string(this.module.GetType()), "equip", &pb.EquipmentEquipResp{Equipments: equipments})
 	return
 }

modules/equipment/api_getlist.go

@@ -20,15 +20,11 @@ func (this *apiComp) Getlist(session comm.IUserSession, req *pb.EquipmentGetList
 		err   error
 		items []*pb.DB_Equipment
 	)
-	defer func() {
-		if code == pb.ErrorCode_Success {
-			session.SendMsg(string(this.module.GetType()), "getlist", &pb.EquipmentGetListResp{Equipments: items})
-		}
-	}()
 	if items, err = this.module.modelEquipment.QueryUserEquipments(session.GetUserId()); err != nil {
 		log.Errorf("QueryUserPackReq err:%v", err)
 		code = pb.ErrorCode_CacheReadError
 		return
 	}
+	session.SendMsg(string(this.module.GetType()), "getlist", &pb.EquipmentGetListResp{Equipments: items})
 	return
 }

modules/equipment/api_upgrade.go

@@ -35,11 +35,6 @@ func (this *apiComp) Upgrade(session comm.IUserSession, req *pb.EquipmentUpgrade
 		equipments       []*pb.DB_Equipment
 		issucc           bool
 	)
-	defer func() {
-		if code == pb.ErrorCode_Success {
-			session.SendMsg(string(this.module.GetType()), "upgrade", &pb.EquipmentUpgradeResp{IsSucc: issucc, Equipment: modifyequipments})
-		}
-	}()
 	if code = this.UpgradeCheck(session, req); code != pb.ErrorCode_Success {
 		return
 	}
@@ -145,5 +140,6 @@ func (this *apiComp) Upgrade(session comm.IUserSession, req *pb.EquipmentUpgrade
 			code = this.module.ModuleHero.UpdateEquipment(hero, equipments)
 		}
 	}
+	session.SendMsg(string(this.module.GetType()), "upgrade", &pb.EquipmentUpgradeResp{IsSucc: issucc, Equipment: modifyequipments})
 	return
 }

modules/equipment/modelEquipment.go

@@ -42,7 +42,9 @@ func (this *modelEquipmentComp) QueryUserEquipmentsById(uId, id string) (equipme
 ///查询用户的武器背包
 func (this *modelEquipmentComp) QueryUserEquipments(uId string) (equipments []*pb.DB_Equipment, err error) {
 	equipments = make([]*pb.DB_Equipment, 0)
-	err = this.GetList(uId, &equipments)
+	if err = this.GetList(uId, &equipments); err != nil {
+		this.module.Errorf("err:%v", err)
+	}
 	return
 }
 

modules/forum/module_test.go

@@ -1,4 +1,4 @@
-package forum
+package forum_test
 
 import (
 	"fmt"
@@ -7,6 +7,7 @@ import (
 	"go_dreamfactory/lego/base/rpcx"
 	"go_dreamfactory/lego/core"
 	"go_dreamfactory/lego/sys/log"
+	"go_dreamfactory/modules/forum"
 	"go_dreamfactory/services"
 	"go_dreamfactory/sys/cache"
 	"go_dreamfactory/sys/configure"
@@ -49,7 +50,7 @@ func (this *TestService) InitSys() {
 
 var service core.IService
 var s_gateComp comm.ISC_GateRouteComp = services.NewGateRouteComp()
-var module = new(Forum)
+var module = new(forum.Forum)
 
 //测试环境下初始化db和cache 系统
 func TestMain(m *testing.M) {

pb.bat

@@ -18,4 +18,5 @@ protoc --proto_path=%SRC%  --go_out=%TAR% --go_opt=paths=import %SRC%\hero\*.pro
 protoc --proto_path=%SRC%  --go_out=%TAR% --go_opt=paths=import %SRC%\shop\*.proto
 protoc --proto_path=%SRC%  --go_out=%TAR% --go_opt=paths=import %SRC%\story\*.proto
 protoc --proto_path=%SRC%  --go_out=%TAR% --go_opt=paths=import %SRC%\task\*.proto
+protoc --proto_path=%SRC%  --go_out=%TAR% --go_opt=paths=import %SRC%\chat\*.proto
 pause

pb/chat_db.pb.go

@@ -0,0 +1,290 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.28.0
+// 	protoc        v3.20.0
+// source: chat/chat_db.proto
+
+package pb
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type ChatChannel int32
+
+const (
+	ChatChannel_World       ChatChannel = 0 //世界频道
+	ChatChannel_Union       ChatChannel = 1 //工会频道
+	ChatChannel_Private     ChatChannel = 2 //私有频道
+	ChatChannel_CrossServer ChatChannel = 3 //跨服频道
+	ChatChannel_System      ChatChannel = 4 //系统频道
+)
+
+// Enum value maps for ChatChannel.
+var (
+	ChatChannel_name = map[int32]string{
+		0: "World",
+		1: "Union",
+		2: "Private",
+		3: "CrossServer",
+		4: "System",
+	}
+	ChatChannel_value = map[string]int32{
+		"World":       0,
+		"Union":       1,
+		"Private":     2,
+		"CrossServer": 3,
+		"System":      4,
+	}
+)
+
+func (x ChatChannel) Enum() *ChatChannel {
+	p := new(ChatChannel)
+	*p = x
+	return p
+}
+
+func (x ChatChannel) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (ChatChannel) Descriptor() protoreflect.EnumDescriptor {
+	return file_chat_chat_db_proto_enumTypes[0].Descriptor()
+}
+
+func (ChatChannel) Type() protoreflect.EnumType {
+	return &file_chat_chat_db_proto_enumTypes[0]
+}
+
+func (x ChatChannel) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use ChatChannel.Descriptor instead.
+func (ChatChannel) EnumDescriptor() ([]byte, []int) {
+	return file_chat_chat_db_proto_rawDescGZIP(), []int{0}
+}
+
+type DBChat struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Id      string      `protobuf:"bytes,1,opt,name=id,proto3" json:"id"`                             //主键id
+	Channel ChatChannel `protobuf:"varint,2,opt,name=channel,proto3,enum=ChatChannel" json:"channel"` //频道
+	Suid    string      `protobuf:"bytes,3,opt,name=suid,proto3" json:"suid"`                         //发送用户id
+	Ruid    string      `protobuf:"bytes,4,opt,name=ruid,proto3" json:"ruid"`                         //接收用户id channel == Private 有效
+	AreaId  string      `protobuf:"bytes,5,opt,name=areaId,proto3" json:"areaId"`                     //区服id
+	UnionId string      `protobuf:"bytes,6,opt,name=unionId,proto3" json:"unionId"`                   //工会id
+	Headid  int32       `protobuf:"varint,7,opt,name=headid,proto3" json:"headid"`                    //用户头像
+	Uname   string      `protobuf:"bytes,8,opt,name=uname,proto3" json:"uname"`                       //用户名
+	Content string      `protobuf:"bytes,9,opt,name=content,proto3" json:"content"`                   //内容
+	Ctime   int64       `protobuf:"varint,10,opt,name=ctime,proto3" json:"ctime"`                     //创建时间
+}
+
+func (x *DBChat) Reset() {
+	*x = DBChat{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_db_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *DBChat) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*DBChat) ProtoMessage() {}
+
+func (x *DBChat) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_db_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use DBChat.ProtoReflect.Descriptor instead.
+func (*DBChat) Descriptor() ([]byte, []int) {
+	return file_chat_chat_db_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *DBChat) GetId() string {
+	if x != nil {
+		return x.Id
+	}
+	return ""
+}
+
+func (x *DBChat) GetChannel() ChatChannel {
+	if x != nil {
+		return x.Channel
+	}
+	return ChatChannel_World
+}
+
+func (x *DBChat) GetSuid() string {
+	if x != nil {
+		return x.Suid
+	}
+	return ""
+}
+
+func (x *DBChat) GetRuid() string {
+	if x != nil {
+		return x.Ruid
+	}
+	return ""
+}
+
+func (x *DBChat) GetAreaId() string {
+	if x != nil {
+		return x.AreaId
+	}
+	return ""
+}
+
+func (x *DBChat) GetUnionId() string {
+	if x != nil {
+		return x.UnionId
+	}
+	return ""
+}
+
+func (x *DBChat) GetHeadid() int32 {
+	if x != nil {
+		return x.Headid
+	}
+	return 0
+}
+
+func (x *DBChat) GetUname() string {
+	if x != nil {
+		return x.Uname
+	}
+	return ""
+}
+
+func (x *DBChat) GetContent() string {
+	if x != nil {
+		return x.Content
+	}
+	return ""
+}
+
+func (x *DBChat) GetCtime() int64 {
+	if x != nil {
+		return x.Ctime
+	}
+	return 0
+}
+
+var File_chat_chat_db_proto protoreflect.FileDescriptor
+
+var file_chat_chat_db_proto_rawDesc = []byte{
+	0x0a, 0x12, 0x63, 0x68, 0x61, 0x74, 0x2f, 0x63, 0x68, 0x61, 0x74, 0x5f, 0x64, 0x62, 0x2e, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x22, 0xf8, 0x01, 0x0a, 0x06, 0x44, 0x42, 0x43, 0x68, 0x61, 0x74, 0x12,
+	0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12,
+	0x26, 0x0a, 0x07, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e,
+	0x32, 0x0c, 0x2e, 0x43, 0x68, 0x61, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x52, 0x07,
+	0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x12, 0x12, 0x0a, 0x04, 0x73, 0x75, 0x69, 0x64, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x73, 0x75, 0x69, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x72,
+	0x75, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x72, 0x75, 0x69, 0x64, 0x12,
+	0x16, 0x0a, 0x06, 0x61, 0x72, 0x65, 0x61, 0x49, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x06, 0x61, 0x72, 0x65, 0x61, 0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x75, 0x6e, 0x69, 0x6f, 0x6e,
+	0x49, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x75, 0x6e, 0x69, 0x6f, 0x6e, 0x49,
+	0x64, 0x12, 0x16, 0x0a, 0x06, 0x68, 0x65, 0x61, 0x64, 0x69, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28,
+	0x05, 0x52, 0x06, 0x68, 0x65, 0x61, 0x64, 0x69, 0x64, 0x12, 0x14, 0x0a, 0x05, 0x75, 0x6e, 0x61,
+	0x6d, 0x65, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x75, 0x6e, 0x61, 0x6d, 0x65, 0x12,
+	0x18, 0x0a, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x63, 0x74, 0x69,
+	0x6d, 0x65, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x63, 0x74, 0x69, 0x6d, 0x65, 0x2a,
+	0x4d, 0x0a, 0x0b, 0x43, 0x68, 0x61, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x12, 0x09,
+	0x0a, 0x05, 0x57, 0x6f, 0x72, 0x6c, 0x64, 0x10, 0x00, 0x12, 0x09, 0x0a, 0x05, 0x55, 0x6e, 0x69,
+	0x6f, 0x6e, 0x10, 0x01, 0x12, 0x0b, 0x0a, 0x07, 0x50, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x10,
+	0x02, 0x12, 0x0f, 0x0a, 0x0b, 0x43, 0x72, 0x6f, 0x73, 0x73, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72,
+	0x10, 0x03, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x79, 0x73, 0x74, 0x65, 0x6d, 0x10, 0x04, 0x42, 0x06,
+	0x5a, 0x04, 0x2e, 0x3b, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_chat_chat_db_proto_rawDescOnce sync.Once
+	file_chat_chat_db_proto_rawDescData = file_chat_chat_db_proto_rawDesc
+)
+
+func file_chat_chat_db_proto_rawDescGZIP() []byte {
+	file_chat_chat_db_proto_rawDescOnce.Do(func() {
+		file_chat_chat_db_proto_rawDescData = protoimpl.X.CompressGZIP(file_chat_chat_db_proto_rawDescData)
+	})
+	return file_chat_chat_db_proto_rawDescData
+}
+
+var file_chat_chat_db_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_chat_chat_db_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_chat_chat_db_proto_goTypes = []interface{}{
+	(ChatChannel)(0), // 0: ChatChannel
+	(*DBChat)(nil),   // 1: DBChat
+}
+var file_chat_chat_db_proto_depIdxs = []int32{
+	0, // 0: DBChat.channel:type_name -> ChatChannel
+	1, // [1:1] is the sub-list for method output_type
+	1, // [1:1] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_chat_chat_db_proto_init() }
+func file_chat_chat_db_proto_init() {
+	if File_chat_chat_db_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_chat_chat_db_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*DBChat); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_chat_chat_db_proto_rawDesc,
+			NumEnums:      1,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_chat_chat_db_proto_goTypes,
+		DependencyIndexes: file_chat_chat_db_proto_depIdxs,
+		EnumInfos:         file_chat_chat_db_proto_enumTypes,
+		MessageInfos:      file_chat_chat_db_proto_msgTypes,
+	}.Build()
+	File_chat_chat_db_proto = out.File
+	file_chat_chat_db_proto_rawDesc = nil
+	file_chat_chat_db_proto_goTypes = nil
+	file_chat_chat_db_proto_depIdxs = nil
+}

pb/chat_msg.pb.go

@@ -0,0 +1,532 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.28.0
+// 	protoc        v3.20.0
+// source: chat/chat_msg.proto
+
+package pb
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+//聊天消息推送
+type ChatMessagePush struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Chats []*DBChat `protobuf:"bytes,1,rep,name=Chats,proto3" json:"Chats"`
+}
+
+func (x *ChatMessagePush) Reset() {
+	*x = ChatMessagePush{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_msg_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ChatMessagePush) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChatMessagePush) ProtoMessage() {}
+
+func (x *ChatMessagePush) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_msg_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChatMessagePush.ProtoReflect.Descriptor instead.
+func (*ChatMessagePush) Descriptor() ([]byte, []int) {
+	return file_chat_chat_msg_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *ChatMessagePush) GetChats() []*DBChat {
+	if x != nil {
+		return x.Chats
+	}
+	return nil
+}
+
+//请求未读消息
+type ChatGetListReq struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *ChatGetListReq) Reset() {
+	*x = ChatGetListReq{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_msg_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ChatGetListReq) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChatGetListReq) ProtoMessage() {}
+
+func (x *ChatGetListReq) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_msg_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChatGetListReq.ProtoReflect.Descriptor instead.
+func (*ChatGetListReq) Descriptor() ([]byte, []int) {
+	return file_chat_chat_msg_proto_rawDescGZIP(), []int{1}
+}
+
+type ChatGetListResp struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Chats []*DBChat `protobuf:"bytes,1,rep,name=Chats,proto3" json:"Chats"`
+}
+
+func (x *ChatGetListResp) Reset() {
+	*x = ChatGetListResp{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_msg_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ChatGetListResp) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChatGetListResp) ProtoMessage() {}
+
+func (x *ChatGetListResp) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_msg_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChatGetListResp.ProtoReflect.Descriptor instead.
+func (*ChatGetListResp) Descriptor() ([]byte, []int) {
+	return file_chat_chat_msg_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *ChatGetListResp) GetChats() []*DBChat {
+	if x != nil {
+		return x.Chats
+	}
+	return nil
+}
+
+//消息发送请求
+type ChatSendReq struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Channel  ChatChannel `protobuf:"varint,1,opt,name=channel,proto3,enum=ChatChannel" json:"channel"` //频道
+	TargetId string      `protobuf:"bytes,2,opt,name=targetId,proto3" json:"targetId"`                 //目标用户id
+	Content  string      `protobuf:"bytes,3,opt,name=content,proto3" json:"content"`                   //内容
+}
+
+func (x *ChatSendReq) Reset() {
+	*x = ChatSendReq{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_msg_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ChatSendReq) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChatSendReq) ProtoMessage() {}
+
+func (x *ChatSendReq) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_msg_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChatSendReq.ProtoReflect.Descriptor instead.
+func (*ChatSendReq) Descriptor() ([]byte, []int) {
+	return file_chat_chat_msg_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *ChatSendReq) GetChannel() ChatChannel {
+	if x != nil {
+		return x.Channel
+	}
+	return ChatChannel_World
+}
+
+func (x *ChatSendReq) GetTargetId() string {
+	if x != nil {
+		return x.TargetId
+	}
+	return ""
+}
+
+func (x *ChatSendReq) GetContent() string {
+	if x != nil {
+		return x.Content
+	}
+	return ""
+}
+
+//消息发送请求 回应
+type ChatSendResp struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *ChatSendResp) Reset() {
+	*x = ChatSendResp{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_msg_proto_msgTypes[4]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ChatSendResp) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChatSendResp) ProtoMessage() {}
+
+func (x *ChatSendResp) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_msg_proto_msgTypes[4]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChatSendResp.ProtoReflect.Descriptor instead.
+func (*ChatSendResp) Descriptor() ([]byte, []int) {
+	return file_chat_chat_msg_proto_rawDescGZIP(), []int{4}
+}
+
+//跨服消息发送请求
+type ChatSpanSendReq struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Channel ChatChannel `protobuf:"varint,1,opt,name=channel,proto3,enum=ChatChannel" json:"channel"` //频道
+	Content string      `protobuf:"bytes,2,opt,name=content,proto3" json:"content"`                   //内容
+}
+
+func (x *ChatSpanSendReq) Reset() {
+	*x = ChatSpanSendReq{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_msg_proto_msgTypes[5]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ChatSpanSendReq) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChatSpanSendReq) ProtoMessage() {}
+
+func (x *ChatSpanSendReq) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_msg_proto_msgTypes[5]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChatSpanSendReq.ProtoReflect.Descriptor instead.
+func (*ChatSpanSendReq) Descriptor() ([]byte, []int) {
+	return file_chat_chat_msg_proto_rawDescGZIP(), []int{5}
+}
+
+func (x *ChatSpanSendReq) GetChannel() ChatChannel {
+	if x != nil {
+		return x.Channel
+	}
+	return ChatChannel_World
+}
+
+func (x *ChatSpanSendReq) GetContent() string {
+	if x != nil {
+		return x.Content
+	}
+	return ""
+}
+
+//跨服消息发送请求 回应
+type ChatSpanSendResp struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *ChatSpanSendResp) Reset() {
+	*x = ChatSpanSendResp{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_chat_chat_msg_proto_msgTypes[6]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ChatSpanSendResp) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChatSpanSendResp) ProtoMessage() {}
+
+func (x *ChatSpanSendResp) ProtoReflect() protoreflect.Message {
+	mi := &file_chat_chat_msg_proto_msgTypes[6]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChatSpanSendResp.ProtoReflect.Descriptor instead.
+func (*ChatSpanSendResp) Descriptor() ([]byte, []int) {
+	return file_chat_chat_msg_proto_rawDescGZIP(), []int{6}
+}
+
+var File_chat_chat_msg_proto protoreflect.FileDescriptor
+
+var file_chat_chat_msg_proto_rawDesc = []byte{
+	0x0a, 0x13, 0x63, 0x68, 0x61, 0x74, 0x2f, 0x63, 0x68, 0x61, 0x74, 0x5f, 0x6d, 0x73, 0x67, 0x2e,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x12, 0x63, 0x68, 0x61, 0x74, 0x2f, 0x63, 0x68, 0x61, 0x74,
+	0x5f, 0x64, 0x62, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x30, 0x0a, 0x0f, 0x43, 0x68, 0x61,
+	0x74, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x50, 0x75, 0x73, 0x68, 0x12, 0x1d, 0x0a, 0x05,
+	0x43, 0x68, 0x61, 0x74, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x44, 0x42,
+	0x43, 0x68, 0x61, 0x74, 0x52, 0x05, 0x43, 0x68, 0x61, 0x74, 0x73, 0x22, 0x10, 0x0a, 0x0e, 0x43,
+	0x68, 0x61, 0x74, 0x47, 0x65, 0x74, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x71, 0x22, 0x30, 0x0a,
+	0x0f, 0x43, 0x68, 0x61, 0x74, 0x47, 0x65, 0x74, 0x4c, 0x69, 0x73, 0x74, 0x52, 0x65, 0x73, 0x70,
+	0x12, 0x1d, 0x0a, 0x05, 0x43, 0x68, 0x61, 0x74, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32,
+	0x07, 0x2e, 0x44, 0x42, 0x43, 0x68, 0x61, 0x74, 0x52, 0x05, 0x43, 0x68, 0x61, 0x74, 0x73, 0x22,
+	0x6b, 0x0a, 0x0b, 0x43, 0x68, 0x61, 0x74, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12, 0x26,
+	0x0a, 0x07, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32,
+	0x0c, 0x2e, 0x43, 0x68, 0x61, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x52, 0x07, 0x63,
+	0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x12, 0x1a, 0x0a, 0x08, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74,
+	0x49, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x74, 0x61, 0x72, 0x67, 0x65, 0x74,
+	0x49, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x18, 0x03, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x22, 0x0e, 0x0a, 0x0c,
+	0x43, 0x68, 0x61, 0x74, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x73, 0x70, 0x22, 0x53, 0x0a, 0x0f,
+	0x43, 0x68, 0x61, 0x74, 0x53, 0x70, 0x61, 0x6e, 0x53, 0x65, 0x6e, 0x64, 0x52, 0x65, 0x71, 0x12,
+	0x26, 0x0a, 0x07, 0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e,
+	0x32, 0x0c, 0x2e, 0x43, 0x68, 0x61, 0x74, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x52, 0x07,
+	0x63, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x12, 0x18, 0x0a, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65,
+	0x6e, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e,
+	0x74, 0x22, 0x12, 0x0a, 0x10, 0x43, 0x68, 0x61, 0x74, 0x53, 0x70, 0x61, 0x6e, 0x53, 0x65, 0x6e,
+	0x64, 0x52, 0x65, 0x73, 0x70, 0x42, 0x06, 0x5a, 0x04, 0x2e, 0x3b, 0x70, 0x62, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_chat_chat_msg_proto_rawDescOnce sync.Once
+	file_chat_chat_msg_proto_rawDescData = file_chat_chat_msg_proto_rawDesc
+)
+
+func file_chat_chat_msg_proto_rawDescGZIP() []byte {
+	file_chat_chat_msg_proto_rawDescOnce.Do(func() {
+		file_chat_chat_msg_proto_rawDescData = protoimpl.X.CompressGZIP(file_chat_chat_msg_proto_rawDescData)
+	})
+	return file_chat_chat_msg_proto_rawDescData
+}
+
+var file_chat_chat_msg_proto_msgTypes = make([]protoimpl.MessageInfo, 7)
+var file_chat_chat_msg_proto_goTypes = []interface{}{
+	(*ChatMessagePush)(nil),  // 0: ChatMessagePush
+	(*ChatGetListReq)(nil),   // 1: ChatGetListReq
+	(*ChatGetListResp)(nil),  // 2: ChatGetListResp
+	(*ChatSendReq)(nil),      // 3: ChatSendReq
+	(*ChatSendResp)(nil),     // 4: ChatSendResp
+	(*ChatSpanSendReq)(nil),  // 5: ChatSpanSendReq
+	(*ChatSpanSendResp)(nil), // 6: ChatSpanSendResp
+	(*DBChat)(nil),           // 7: DBChat
+	(ChatChannel)(0),         // 8: ChatChannel
+}
+var file_chat_chat_msg_proto_depIdxs = []int32{
+	7, // 0: ChatMessagePush.Chats:type_name -> DBChat
+	7, // 1: ChatGetListResp.Chats:type_name -> DBChat
+	8, // 2: ChatSendReq.channel:type_name -> ChatChannel
+	8, // 3: ChatSpanSendReq.channel:type_name -> ChatChannel
+	4, // [4:4] is the sub-list for method output_type
+	4, // [4:4] is the sub-list for method input_type
+	4, // [4:4] is the sub-list for extension type_name
+	4, // [4:4] is the sub-list for extension extendee
+	0, // [0:4] is the sub-list for field type_name
+}
+
+func init() { file_chat_chat_msg_proto_init() }
+func file_chat_chat_msg_proto_init() {
+	if File_chat_chat_msg_proto != nil {
+		return
+	}
+	file_chat_chat_db_proto_init()
+	if !protoimpl.UnsafeEnabled {
+		file_chat_chat_msg_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ChatMessagePush); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_chat_chat_msg_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ChatGetListReq); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_chat_chat_msg_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ChatGetListResp); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_chat_chat_msg_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ChatSendReq); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_chat_chat_msg_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ChatSendResp); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_chat_chat_msg_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ChatSpanSendReq); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_chat_chat_msg_proto_msgTypes[6].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ChatSpanSendResp); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_chat_chat_msg_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   7,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_chat_chat_msg_proto_goTypes,
+		DependencyIndexes: file_chat_chat_msg_proto_depIdxs,
+		MessageInfos:      file_chat_chat_msg_proto_msgTypes,
+	}.Build()
+	File_chat_chat_msg_proto = out.File
+	file_chat_chat_msg_proto_rawDesc = nil
+	file_chat_chat_msg_proto_goTypes = nil
+	file_chat_chat_msg_proto_depIdxs = nil
+}

pb/proto/chat/chat_db.proto

@@ -0,0 +1,24 @@
+syntax = "proto3";
+option go_package = ".;pb";
+
+enum ChatChannel {
+    World = 0;              //世界频道
+    Union = 1;              //工会频道
+    Private = 2;            //私有频道
+    CrossServer = 3;        //跨服频道
+    System = 4;             //系统频道
+}
+
+
+message DBChat {
+    string id =1;               //主键id
+    ChatChannel channel = 2;    //频道
+    string suid =3;             //发送用户id
+    string ruid = 4;            //接收用户id channel == Private 有效
+    string areaId = 5;          //区服id
+    string unionId = 6;         //工会id
+    int32 headid = 7;           //用户头像
+    string uname = 8;           //用户名
+    string content = 9;         //内容
+    int64 ctime = 10;            //创建时间 
+}

pb/proto/chat/chat_msg.proto

@@ -0,0 +1,39 @@
+syntax = "proto3";
+option go_package = ".;pb";
+import "chat/chat_db.proto";
+
+//聊天消息推送
+message ChatMessagePush{
+    repeated DBChat Chats = 1;
+}
+
+//请求未读消息
+message ChatGetListReq {
+
+}
+message ChatGetListResp {
+    repeated DBChat Chats = 1;
+}
+
+//消息发送请求
+message ChatSendReq {
+    ChatChannel channel = 1;    //频道
+    string targetId = 2;        //目标用户id
+    string content = 3;         //内容
+}
+//消息发送请求 回应
+message ChatSendResp {
+
+}
+
+
+//跨服消息发送请求
+message ChatSpanSendReq {
+    ChatChannel channel = 1;    //频道
+    string content = 2;         //内容
+}
+//跨服消息发送请求 回应
+message ChatSpanSendResp {
+
+}
+