msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

rpc start

Browse Source
test-errcode
withchao 3 years ago
parent f3d43f8578
commit 048d4b83e1
5 changed files with 206 additions and 112 deletions
Show Stats Download Patch File Download Diff File

34
cmd/rpc/group/main.go
Unescape View File

@ -2,25 +2,25 @@ package main
import ( import (
"Open_IM/internal/rpc/group" "Open_IM/internal/rpc/group"
"Open_IM/internal/startrpc"
"Open_IM/pkg/common/config" "Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
promePkg "Open_IM/pkg/common/prometheus"
"flag"
"fmt"
) )
func main() { func main() {
defaultPorts := config.Config.RpcPort.OpenImGroupPort //defaultPorts := config.Config.RpcPort.OpenImGroupPort
rpcPort := flag.Int("port", defaultPorts[0], "get RpcGroupPort from cmd,default 16000 as port") //rpcPort := flag.Int("port", defaultPorts[0], "get RpcGroupPort from cmd,default 16000 as port")
prometheusPort := flag.Int("prometheus_port", config.Config.Prometheus.GroupPrometheusPort[0], "groupPrometheusPort default listen port") //prometheusPort := flag.Int("prometheus_port", config.Config.Prometheus.GroupPrometheusPort[0], "groupPrometheusPort default listen port")
flag.Parse() //flag.Parse()
fmt.Println("start group rpc server, port: ", *rpcPort, ", OpenIM version: ", constant.CurrentVersion, "\n") //fmt.Println("start group rpc server, port: ", *rpcPort, ", OpenIM version: ", constant.CurrentVersion, "\n")
rpcServer := group.NewGroupServer(*rpcPort) //rpcServer := group.NewGroupServer(*rpcPort)
go func() { //go func() {
err := promePkg.StartPromeSrv(*prometheusPort) // err := promePkg.StartPromeSrv(*prometheusPort)
if err != nil { // if err != nil {
panic(err) // panic(err)
} // }
}() //}()
rpcServer.Run() //rpcServer.Run()
startrpc.StartRpc(config.Config.RpcPort.OpenImGroupPort[0], config.Config.RpcRegisterName.OpenImGroupName, config.Config.Prometheus.GroupPrometheusPort[0], group.Start)
} }

206
internal/rpc/group/group.go
Unescape View File

@ -2,18 +2,13 @@ package group
import ( import (
"Open_IM/internal/common/check" "Open_IM/internal/common/check"
"Open_IM/internal/common/network"
chat "Open_IM/internal/rpc/msg" chat "Open_IM/internal/rpc/msg"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db/cache" "Open_IM/pkg/common/db/cache"
"Open_IM/pkg/common/db/controller" "Open_IM/pkg/common/db/controller"
"Open_IM/pkg/common/db/relation" "Open_IM/pkg/common/db/relation"
relationTb "Open_IM/pkg/common/db/table/relation" relationTb "Open_IM/pkg/common/db/table/relation"
"Open_IM/pkg/common/db/unrelation" "Open_IM/pkg/common/db/unrelation"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/middleware"
promePkg "Open_IM/pkg/common/prometheus"
"Open_IM/pkg/common/tokenverify" "Open_IM/pkg/common/tokenverify"
"Open_IM/pkg/common/tracelog" "Open_IM/pkg/common/tracelog"
discoveryRegistry "Open_IM/pkg/discoveryregistry" discoveryRegistry "Open_IM/pkg/discoveryregistry"
@ -22,107 +17,128 @@ import (
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
"fmt" "fmt"
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus" "google.golang.org/grpc"
"gorm.io/gorm" "gorm.io/gorm"
"math/big" "math/big"
"math/rand" "math/rand"
"net"
"strconv" "strconv"
"strings" "strings"
"time" "time"
"github.com/OpenIMSDK/openKeeper"
"google.golang.org/grpc"
) )
type groupServer struct { func Start(server *grpc.Server) {
rpcPort int //err := mysql.InitConn().AutoMigrateModel(&groupModel)
rpcRegisterName string //if err != nil {
schema string // panic("db init err:" + err.Error())
zkAddr []string //}
GroupInterface controller.GroupInterface //mongo.InitMongo()
registerCenter discoveryRegistry.SvcDiscoveryRegistry //redis.InitRedis()
user *check.UserCheck pbGroup.RegisterGroupServer(server, &groupServer{
GroupInterface: controller.NewGroupInterface(nil, cache.NewRedis().GetClient(), unrelation.NewMongo().GetClient()),
registerCenter: nil,
user: check.NewUserCheck(),
})
} }
func NewGroupServer(port int) *groupServer { type groupServer struct {
log.NewPrivateLog(constant.LogFileName) //rpcPort int
g := groupServer{ //rpcRegisterName string
rpcPort: port, //schema string
rpcRegisterName: config.Config.RpcRegisterName.OpenImGroupName, //zkAddr []string
schema: config.Config.Zookeeper.Schema, GroupInterface controller.GroupInterface
zkAddr: config.Config.Zookeeper.ZkAddr, registerCenter discoveryRegistry.SvcDiscoveryRegistry
} user *check.UserCheck
//mysql init
var mysql relation.Mysql
var mongo unrelation.Mongo
var groupModel relationTb.GroupModel
var redis cache.RedisClient
err := mysql.InitConn().AutoMigrateModel(&groupModel)
if err != nil {
panic("db init err:" + err.Error())
}
mongo.InitMongo()
redis.InitRedis()
mongo.CreateSuperGroupIndex()
zkClient, err := openKeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema, 10, "", "")
if err != nil {
panic(err.Error())
}
registerIP, err := network.GetRpcRegisterIP(config.Config.RpcRegisterIP)
g.registerCenter = zkClient
err = g.registerCenter.Register(config.Config.RpcRegisterName.OpenImGroupName, registerIP, port)
if err != nil {
panic(err.Error())
}
//conns, err := g.registerCenter.GetConns(config.Config.RpcRegisterName.OpenImConversationName)
g.GroupInterface = controller.NewGroupInterface(mysql.GormConn(), redis.GetClient(), mongo.GetClient())
g.user = check.NewUserCheck()
return &g
} }
func (s *groupServer) Run() { //
operationID := utils.OperationIDGenerator() //type groupServer struct {
log.NewInfo(operationID, "group rpc start ") // rpcPort int
address := network.GetListenIP(config.Config.ListenIP) + ":" + strconv.Itoa(s.rpcPort) // rpcRegisterName string
//listener network // schema string
listener, err := net.Listen("tcp", address) // zkAddr []string
if err != nil { // GroupInterface controller.GroupInterface
panic("listening err:" + err.Error() + s.rpcRegisterName) // registerCenter discoveryRegistry.SvcDiscoveryRegistry
} // user *check.UserCheck
log.NewInfo(operationID, "listen network success, ", address, listener) //}
//
defer listener.Close() //func NewGroupServer(port int) *groupServer {
//grpc server // log.NewPrivateLog(constant.LogFileName)
recvSize := 1024 * 1024 * constant.GroupRPCRecvSize // g := groupServer{
sendSize := 1024 * 1024 * constant.GroupRPCSendSize // rpcPort: port,
var grpcOpts = []grpc.ServerOption{ // rpcRegisterName: config.Config.RpcRegisterName.OpenImGroupName,
grpc.MaxRecvMsgSize(recvSize), // schema: config.Config.Zookeeper.Schema,
grpc.MaxSendMsgSize(sendSize), // zkAddr: config.Config.Zookeeper.ZkAddr,
grpc.UnaryInterceptor(middleware.RpcServerInterceptor), // }
} // //mysql init
if config.Config.Prometheus.Enable { // var mysql relation.Mysql
promePkg.NewGrpcRequestCounter() // var mongo unrelation.Mongo
promePkg.NewGrpcRequestFailedCounter() // var groupModel relationTb.GroupModel
promePkg.NewGrpcRequestSuccessCounter() // var redis cache.RedisClient
grpcOpts = append(grpcOpts, []grpc.ServerOption{ // err := mysql.InitConn().AutoMigrateModel(&groupModel)
// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme), // if err != nil {
grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor), // panic("db init err:" + err.Error())
grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor), // }
}...) // mongo.InitMongo()
} // redis.InitRedis()
srv := grpc.NewServer(grpcOpts...) // mongo.CreateSuperGroupIndex()
defer srv.GracefulStop() // zkClient, err := openKeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema, 10, "", "")
//Service registers with etcd // if err != nil {
pbGroup.RegisterGroupServer(srv, s) // panic(err.Error())
err = srv.Serve(listener) // }
if err != nil { // registerIP, err := network.GetRpcRegisterIP(config.Config.RpcRegisterIP)
log.NewError(operationID, "Serve failed ", err.Error()) // g.registerCenter = zkClient
return // err = g.registerCenter.Register(config.Config.RpcRegisterName.OpenImGroupName, registerIP, port)
} // if err != nil {
log.NewInfo(operationID, "group rpc success") // panic(err.Error())
} // }
//
// //conns, err := g.registerCenter.GetConns(config.Config.RpcRegisterName.OpenImConversationName)
// g.GroupInterface = controller.NewGroupInterface(mysql.GormConn(), redis.GetClient(), mongo.GetClient())
// g.user = check.NewUserCheck()
// return &g
//}
//
//func (s *groupServer) Run() {
// operationID := utils.OperationIDGenerator()
// log.NewInfo(operationID, "group rpc start ")
// address := network.GetListenIP(config.Config.ListenIP) + ":" + strconv.Itoa(s.rpcPort)
// //listener network
// listener, err := net.Listen("tcp", address)
// if err != nil {
// panic("listening err:" + err.Error() + s.rpcRegisterName)
// }
// log.NewInfo(operationID, "listen network success, ", address, listener)
//
// defer listener.Close()
// //grpc server
// recvSize := 1024 * 1024 * constant.GroupRPCRecvSize
// sendSize := 1024 * 1024 * constant.GroupRPCSendSize
// var grpcOpts = []grpc.ServerOption{
// grpc.MaxRecvMsgSize(recvSize),
// grpc.MaxSendMsgSize(sendSize),
// grpc.UnaryInterceptor(middleware.RpcServerInterceptor),
// }
// if config.Config.Prometheus.Enable {
// promePkg.NewGrpcRequestCounter()
// promePkg.NewGrpcRequestFailedCounter()
// promePkg.NewGrpcRequestSuccessCounter()
// grpcOpts = append(grpcOpts, []grpc.ServerOption{
// // grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
// grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
// grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
// }...)
// }
// srv := grpc.NewServer(grpcOpts...)
// defer srv.GracefulStop()
// //Service registers with etcd
// pbGroup.RegisterGroupServer(srv, s)
// err = srv.Serve(listener)
// if err != nil {
// log.NewError(operationID, "Serve failed ", err.Error())
// return
// }
// log.NewInfo(operationID, "group rpc success")
//}
func (s *groupServer) CheckGroupAdmin(ctx context.Context, groupID string) error { func (s *groupServer) CheckGroupAdmin(ctx context.Context, groupID string) error {
if !tokenverify.IsAppManagerUid(ctx) { if !tokenverify.IsAppManagerUid(ctx) {
@ -145,7 +161,7 @@ func (s *groupServer) GetUsernameMap(ctx context.Context, userIDs []string, comp
if err != nil { if err != nil {
return nil, err return nil, err
} }
return utils.SliceToMapAny(users, func(e *open_im_sdk.PublicUserInfo) (string, string) { return utils.SliceToMapAny(users, func(e *sdkws.PublicUserInfo) (string, string) {
return e.UserID, e.Nickname return e.UserID, e.Nickname
}), nil }), nil
} }

66
internal/startrpc/start.go
Unescape View File

@ -0,0 +1,66 @@
package startrpc
import (
"Open_IM/internal/common/network"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/middleware"
promePkg "Open_IM/pkg/common/prometheus"
"flag"
"fmt"
"github.com/OpenIMSDK/openKeeper"
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"google.golang.org/grpc"
"net"
)
func StartRpc(rpcPort int, rpcRegisterName string, prometheusPort int, fn func(server *grpc.Server), options ...grpc.ServerOption) {
flagRpcPort := flag.Int("port", rpcPort, "get RpcGroupPort from cmd,default 16000 as port")
flagPrometheusPort := flag.Int("prometheus_port", prometheusPort, "groupPrometheusPort default listen port")
flag.Parse()
rpcPort = *flagRpcPort
prometheusPort = *flagPrometheusPort
fmt.Println("start group rpc server, port: ", rpcPort, ", OpenIM version: ", constant.CurrentVersion)
log.NewPrivateLog(constant.LogFileName)
listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", config.Config.ListenIP, rpcPort))
if err != nil {
panic(err)
}
defer listener.Close()
zkClient, err := openKeeper.NewClient(config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema, 10, "", "")
if err != nil {
panic(err.Error())
}
registerIP, err := network.GetRpcRegisterIP(config.Config.RpcRegisterIP)
if err != nil {
panic(err)
}
options = append(options, grpc.UnaryInterceptor(middleware.RpcServerInterceptor)) // ctx 中间件
if config.Config.Prometheus.Enable {
promePkg.NewGrpcRequestCounter()
promePkg.NewGrpcRequestFailedCounter()
promePkg.NewGrpcRequestSuccessCounter()
options = append(options, []grpc.ServerOption{
// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
}...)
}
srv := grpc.NewServer(options...)
defer srv.GracefulStop()
fn(srv)
err = zkClient.Register(rpcRegisterName, registerIP, rpcPort)
if err != nil {
panic(err.Error())
}
if config.Config.Prometheus.Enable {
err := promePkg.StartPromeSrv(prometheusPort)
if err != nil {
panic(err)
}
}
if err := srv.Serve(listener); err != nil {
panic(err)
}
}

6
pkg/common/db/cache/redis.go vendored
Unescape View File

@ -50,6 +50,12 @@ type Cache interface {
// native redis operate // native redis operate
func NewRedis() *RedisClient {
o := &RedisClient{}
o.InitRedis()
return o
}
type RedisClient struct { type RedisClient struct {
rdb redis.UniversalClient rdb redis.UniversalClient
} }

6
pkg/common/db/unrelation/mongo.go
Unescape View File

@ -13,6 +13,12 @@ import (
"time" "time"
) )
func NewMongo() *Mongo {
mgo := &Mongo{}
mgo.InitMongo()
return mgo
}
type Mongo struct { type Mongo struct {
db *mongo.Client db *mongo.Client
} }

Write Preview
Loading…
Cancel
Save