Merge branch 'errcode' of github.com:OpenIMSDK/Open-IM-Server into errcode

 Conflicts:
	internal/api/group/group.go
	internal/api/group/super_group.go
	pkg/common/db/controller/user.go
This commit is contained in:
wangchuxiao
2023-02-09 14:49:17 +08:00
19 changed files with 1776 additions and 1681 deletions
+1265 -1265
View File
File diff suppressed because it is too large Load Diff
+107
View File
@@ -0,0 +1,107 @@
package group
import (
"Open_IM/pkg/api_struct"
"Open_IM/pkg/proto/group"
"context"
"github.com/gin-gonic/gin"
"google.golang.org/grpc"
"testing"
)
type Ignore struct{}
func temp(client group.GroupClient, ctx context.Context, in *group.KickGroupMemberReq, opts ...grpc.CallOption) (*group.KickGroupMemberResp, error) {
return nil, nil
}
type ApiBind[A, B any] interface {
OperationID() string
OpUserID() (string, error)
Bind(*A) error
Error(error)
Write(*B)
}
func NewApiBind[A, B any](c *gin.Context) ApiBind[A, B] {
return &ginApiBind[A, B]{
c: c,
}
}
type ginApiBind[A, B any] struct {
c *gin.Context
}
func (g *ginApiBind[A, B]) OperationID() string {
return g.c.GetHeader("operationID")
}
func (g *ginApiBind[A, B]) OpUserID() (string, error) {
return "", nil
}
func (g *ginApiBind[A, B]) Bind(a *A) error {
return g.c.BindJSON(a)
}
func (g *ginApiBind[A, B]) Error(err error) {
//TODO implement me
}
func (g *ginApiBind[A, B]) Write(b *B) {
//TODO implement me
}
func TestName(t *testing.T) {
//var bind ApiBind[int, int]
//NewRpc(bind, "", group.NewGroupClient, temp)
var c *gin.Context
NewRpc(NewApiBind[api_struct.KickGroupMemberReq, api_struct.KickGroupMemberResp](c), "", group.NewGroupClient, group.GroupClient.KickGroupMember)
}
func KickGroupMember(c *gin.Context) {
// 默认 全部自动
NewRpc(NewApiBind[api_struct.KickGroupMemberReq, api_struct.KickGroupMemberResp](c), "", group.NewGroupClient, group.GroupClient.KickGroupMember).Execute()
// 可以自定义编辑请求和响应
NewRpc(NewApiBind[api_struct.KickGroupMemberReq, api_struct.KickGroupMemberResp](c), "", group.NewGroupClient, group.GroupClient.KickGroupMember).Before(func(apiReq *api_struct.KickGroupMemberReq, rpcReq *group.KickGroupMemberReq, bind func() error) error {
return bind()
}).After(func(rpcResp *group.KickGroupMemberResp, apiResp *api_struct.KickGroupMemberResp, bind func() error) error {
return bind()
}).Execute()
}
func NewRpc[A, B any, C, D any, Z any](bind ApiBind[A, B], name string, client func(conn *grpc.ClientConn) Z, rpc func(client Z, ctx context.Context, req C, options ...grpc.CallOption) (D, error)) *RpcRun[A, B, C, D, Z] {
return &RpcRun[A, B, C, D, Z]{
bind: bind,
name: name,
client: client,
rpc: rpc,
}
}
type RpcRun[A, B any, C, D any, Z any] struct {
bind ApiBind[A, B]
name string
client func(conn *grpc.ClientConn) Z
rpc func(client Z, ctx context.Context, req C, options ...grpc.CallOption) (D, error)
before func(apiReq *A, rpcReq C, bind func() error) error
after func(rpcResp D, apiResp *B, bind func() error) error
}
func (a *RpcRun[A, B, C, D, Z]) Before(fn func(apiReq *A, rpcReq C, bind func() error) error) *RpcRun[A, B, C, D, Z] {
a.before = fn
return a
}
func (a *RpcRun[A, B, C, D, Z]) After(fn func(rpcResp D, apiResp *B, bind func() error) error) *RpcRun[A, B, C, D, Z] {
a.after = fn
return a
}
func (a *RpcRun[A, B, C, D, Z]) Execute() {
}
+87 -87
View File
@@ -1,89 +1,89 @@
package group
import (
jsonData "Open_IM/internal/utils"
api "Open_IM/pkg/api_struct"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/tokenverify"
"Open_IM/pkg/getcdv3"
rpc "Open_IM/pkg/proto/group"
"Open_IM/pkg/utils"
"context"
"github.com/gin-gonic/gin"
"net/http"
"strings"
)
func GetJoinedSuperGroupList(c *gin.Context) {
req := api.GetJoinedSuperGroupListReq{}
if err := c.BindJSON(&req); err != nil {
log.NewError("0", "BindJSON failed ", err.Error())
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req)
ok, opUserID, errInfo := tokenverify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
reqPb := rpc.GetJoinedSuperGroupListReq{OperationID: req.OperationID, OpUserID: opUserID, UserID: req.FromUserID}
etcdConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName, req.OperationID)
if etcdConn == nil {
errMsg := req.OperationID + "getcdv3.GetDefaultConn == nil"
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
client := rpc.NewGroupClient(etcdConn)
rpcResp, err := client.GetJoinedSuperGroupList(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, "InviteUserToGroup failed ", err.Error(), reqPb.String())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
return
}
GroupListResp := api.GetJoinedSuperGroupListResp{GetJoinedGroupListResp: api.GetJoinedGroupListResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}, GroupInfoList: rpcResp.GroupList}}
GroupListResp.Data = jsonData.JsonDataList(GroupListResp.GroupInfoList)
log.NewInfo(req.OperationID, "GetJoinedSuperGroupList api return ", GroupListResp)
c.JSON(http.StatusOK, GroupListResp)
}
func GetSuperGroupsInfo(c *gin.Context) {
req := api.GetSuperGroupsInfoReq{}
if err := c.BindJSON(&req); err != nil {
log.NewError("0", "BindJSON failed ", err.Error())
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req)
ok, opUserID, errInfo := tokenverify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
if !ok {
errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
reqPb := rpc.GetSuperGroupsInfoReq{OperationID: req.OperationID, OpUserID: opUserID, GroupIDList: req.GroupIDList}
etcdConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName, req.OperationID)
if etcdConn == nil {
errMsg := req.OperationID + "getcdv3.GetDefaultConn == nil"
log.NewError(req.OperationID, errMsg)
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
return
}
client := rpc.NewGroupClient(etcdConn)
rpcResp, err := client.GetSuperGroupsInfo(context.Background(), &reqPb)
if err != nil {
log.NewError(req.OperationID, "InviteUserToGroup failed ", err.Error(), reqPb.String())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
return
}
resp := api.GetSuperGroupsInfoResp{GetGroupInfoResp: api.GetGroupInfoResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}, GroupInfoList: rpcResp.GroupInfoList}}
resp.Data = jsonData.JsonDataList(resp.GroupInfoList)
log.NewInfo(req.OperationID, "GetGroupsInfo api return ", resp)
c.JSON(http.StatusOK, resp)
}
//import (
// jsonData "Open_IM/internal/utils"
// api "Open_IM/pkg/api_struct"
// "Open_IM/pkg/common/config"
// "Open_IM/pkg/common/log"
// "Open_IM/pkg/common/token_verify"
// "Open_IM/pkg/getcdv3"
// rpc "Open_IM/pkg/proto/group"
// "Open_IM/pkg/utils"
// "context"
// "github.com/gin-gonic/gin"
// "net/http"
// "strings"
//)
//
//func GetJoinedSuperGroupList(c *gin.Context) {
// req := api.GetJoinedSuperGroupListReq{}
// if err := c.BindJSON(&req); err != nil {
// log.NewError("0", "BindJSON failed ", err.Error())
// c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
// return
// }
// log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req)
// ok, opUserID, errInfo := token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
// if !ok {
// errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
// log.NewError(req.OperationID, errMsg)
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
// return
// }
// reqPb := rpc.GetJoinedSuperGroupListReq{OperationID: req.OperationID, OpUserID: opUserID, UserID: req.FromUserID}
// etcdConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName, req.OperationID)
// if etcdConn == nil {
// errMsg := req.OperationID + "getcdv3.GetDefaultConn == nil"
// log.NewError(req.OperationID, errMsg)
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
// return
// }
// client := rpc.NewGroupClient(etcdConn)
// rpcResp, err := client.GetJoinedSuperGroupList(context.Background(), &reqPb)
// if err != nil {
// log.NewError(req.OperationID, "InviteUserToGroup failed ", err.Error(), reqPb.String())
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
// return
// }
// GroupListResp := api.GetJoinedSuperGroupListResp{GetJoinedGroupListResp: api.GetJoinedGroupListResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}, GroupInfoList: rpcResp.GroupList}}
// GroupListResp.Data = jsonData.JsonDataList(GroupListResp.GroupInfoList)
// log.NewInfo(req.OperationID, "GetJoinedSuperGroupList api return ", GroupListResp)
// c.JSON(http.StatusOK, GroupListResp)
//}
//
//func GetSuperGroupsInfo(c *gin.Context) {
// req := api.GetSuperGroupsInfoReq{}
// if err := c.BindJSON(&req); err != nil {
// log.NewError("0", "BindJSON failed ", err.Error())
// c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
// return
// }
// log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req)
// ok, opUserID, errInfo := token_verify.GetUserIDFromToken(c.Request.Header.Get("token"), req.OperationID)
// if !ok {
// errMsg := req.OperationID + " " + "GetUserIDFromToken failed " + errInfo + " token:" + c.Request.Header.Get("token")
// log.NewError(req.OperationID, errMsg)
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
// return
// }
// reqPb := rpc.GetSuperGroupsInfoReq{OperationID: req.OperationID, OpUserID: opUserID, GroupIDList: req.GroupIDList}
// etcdConn := rpc.GetDefaultConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName, req.OperationID)
// if etcdConn == nil {
// errMsg := req.OperationID + "getcdv3.GetDefaultConn == nil"
// log.NewError(req.OperationID, errMsg)
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": errMsg})
// return
// }
// client := rpc.NewGroupClient(etcdConn)
// rpcResp, err := client.GetSuperGroupsInfo(context.Background(), &reqPb)
// if err != nil {
// log.NewError(req.OperationID, "InviteUserToGroup failed ", err.Error(), reqPb.String())
// c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
// return
// }
//
// resp := api.GetSuperGroupsInfoResp{GetGroupInfoResp: api.GetGroupInfoResp{CommResp: api.CommResp{ErrCode: rpcResp.CommonResp.ErrCode, ErrMsg: rpcResp.CommonResp.ErrMsg}, GroupInfoList: rpcResp.GroupInfoList}}
// resp.Data = jsonData.JsonDataList(resp.GroupInfoList)
// log.NewInfo(req.OperationID, "GetGroupsInfo api return ", resp)
// c.JSON(http.StatusOK, resp)
//}
+9 -1
View File
@@ -2,7 +2,7 @@ package network
import utils "github.com/OpenIMSDK/open_utils"
func GetRpcIP(configIP string) (string, error) {
func GetRpcRegisterIP(configIP string) (string, error) {
registerIP := configIP
if registerIP == "" {
ip, err := utils.GetLocalIP()
@@ -13,3 +13,11 @@ func GetRpcIP(configIP string) (string, error) {
}
return registerIP, nil
}
func GetListenIP(configIP string) string {
if configIP == "" {
return "0.0.0.0"
} else {
return configIP
}
}
+46
View File
@@ -0,0 +1,46 @@
package rpc_server
import (
"Open_IM/internal/common/network"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log"
discoveryRegistry "Open_IM/pkg/discovery_registry"
"github.com/OpenIMSDK/openKeeper"
"net"
"strconv"
)
type RpcServer struct {
Port int
RegisterName string
RegisterCenter discoveryRegistry.SvcDiscoveryRegistry
}
func NewRpcServer(registerIPInConfig string, port int, registerName string, zkServers []string, zkRoot string) (*RpcServer, error) {
log.NewPrivateLog(constant.LogFileName)
s := &RpcServer{
Port: port,
RegisterName: registerName,
}
zkClient, err := openKeeper.NewClient(zkServers, zkRoot, 10, "", "")
if err != nil {
return nil, err
}
registerIP, err := network.GetRpcRegisterIP(registerIPInConfig)
if err != nil {
return nil, err
}
err = zkClient.Register(s.RegisterName, registerIP, s.Port)
if err != nil {
return nil, err
}
s.RegisterCenter = zkClient
return s, nil
}
func GetTcpListen(listenIPInConfig string, port int) (net.Listener, string, error) {
address := network.GetListenIP(listenIPInConfig) + ":" + strconv.Itoa(port)
listener, err := net.Listen("tcp", address)
return listener, address, err
}
+45 -78
View File
@@ -2,30 +2,63 @@ package auth
import (
"Open_IM/internal/common/check"
"Open_IM/internal/common/network"
"Open_IM/internal/common/rpc_server"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db/controller"
"Open_IM/pkg/common/log"
promePkg "Open_IM/pkg/common/prometheus"
"Open_IM/pkg/common/tokenverify"
"Open_IM/pkg/common/tracelog"
discoveryRegistry "Open_IM/pkg/discovery_registry"
pbAuth "Open_IM/pkg/proto/auth"
pbRelay "Open_IM/pkg/proto/relay"
"Open_IM/pkg/utils"
"context"
"github.com/OpenIMSDK/openKeeper"
"net"
"strconv"
"strings"
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"Open_IM/pkg/common/config"
"google.golang.org/grpc"
)
func NewRpcAuthServer(port int) *rpcAuth {
r, err := rpc_server.NewRpcServer(config.Config.RpcRegisterIP, port, config.Config.RpcRegisterName.OpenImAuthName, config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema)
if err != nil {
panic(err)
}
return &rpcAuth{
RpcServer: r,
}
}
func (s *rpcAuth) Run() {
operationID := utils.OperationIDGenerator()
log.NewInfo(operationID, "rpc auth start...")
listener, address, err := rpc_server.GetTcpListen(config.Config.ListenIP, s.Port)
if err != nil {
panic(err)
}
log.NewInfo(operationID, "listen network success ", listener, address)
var grpcOpts []grpc.ServerOption
if config.Config.Prometheus.Enable {
promePkg.NewGrpcRequestCounter()
promePkg.NewGrpcRequestFailedCounter()
promePkg.NewGrpcRequestSuccessCounter()
promePkg.NewUserRegisterCounter()
promePkg.NewUserLoginCounter()
grpcOpts = append(grpcOpts, []grpc.ServerOption{
// grpc.UnaryInterceptor(promePkg.UnaryServerInterceptorProme),
grpc.StreamInterceptor(grpcPrometheus.StreamServerInterceptor),
grpc.UnaryInterceptor(grpcPrometheus.UnaryServerInterceptor),
}...)
}
srv := grpc.NewServer(grpcOpts...)
defer srv.GracefulStop()
pbAuth.RegisterAuthServer(srv, s)
err = srv.Serve(listener)
if err != nil {
panic(err)
}
log.NewInfo(operationID, "rpc auth ok")
}
func (s *rpcAuth) UserToken(ctx context.Context, req *pbAuth.UserTokenReq) (*pbAuth.UserTokenResp, error) {
resp := pbAuth.UserTokenResp{}
if _, err := check.GetUsersInfo(ctx, req.UserID); err != nil {
@@ -87,7 +120,7 @@ func (s *rpcAuth) ForceLogout(ctx context.Context, req *pbAuth.ForceLogoutReq) (
}
func (s *rpcAuth) forceKickOff(ctx context.Context, userID string, platformID int32, operationID string) error {
grpcCons, err := s.dr.GetConns(config.Config.RpcRegisterName.OpenImRelayName)
grpcCons, err := s.RegisterCenter.GetConns(config.Config.RpcRegisterName.OpenImRelayName)
if err != nil {
return err
}
@@ -102,72 +135,6 @@ func (s *rpcAuth) forceKickOff(ctx context.Context, userID string, platformID in
}
type rpcAuth struct {
rpcPort int
rpcRegisterName string
etcdSchema string
etcdAddr []string
*rpc_server.RpcServer
controller.AuthInterface
dr discoveryRegistry.SvcDiscoveryRegistry
}
func NewRpcAuthServer(port int) *rpcAuth {
log.NewPrivateLog(constant.LogFileName)
return &rpcAuth{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImAuthName,
}
}
func (s *rpcAuth) Run() {
operationID := utils.OperationIDGenerator()
log.NewInfo(operationID, "rpc auth start...")
listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
listener, err := net.Listen("tcp", address)
if err != nil {
panic("listening err:" + err.Error() + s.rpcRegisterName)
}
log.NewInfo(operationID, "listen network success, ", address, listener)
var grpcOpts []grpc.ServerOption
if config.Config.Prometheus.Enable {
promePkg.NewGrpcRequestCounter()
promePkg.NewGrpcRequestFailedCounter()
promePkg.NewGrpcRequestSuccessCounter()
promePkg.NewUserRegisterCounter()
promePkg.NewUserLoginCounter()
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
pbAuth.RegisterAuthServer(srv, s)
zkClient, err := openKeeper.NewClient([]string{"43.154.157.177:2181"}, config.Config.Etcd.EtcdSchema, 10, "", "")
if err != nil {
panic(err.Error())
}
registerIP, err := network.GetRpcIP(config.Config.RpcRegisterIP)
err = zkClient.Register(s.rpcRegisterName, registerIP, s.rpcPort)
if err != nil {
panic(err.Error())
}
log.NewInfo(operationID, "RegisterAuthServer ok ", s.etcdSchema, strings.Join(s.etcdAddr, ","), registerIP, s.rpcPort, s.rpcRegisterName)
err = srv.Serve(listener)
if err != nil {
log.NewError(operationID, "Serve failed ", err.Error())
return
}
log.NewInfo(operationID, "rpc auth ok")
}
+19 -52
View File
@@ -1,7 +1,9 @@
package friend
import (
"Open_IM/internal/common/check"
"Open_IM/internal/common/convert"
"Open_IM/internal/common/rpc_server"
chat "Open_IM/internal/rpc/msg"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
@@ -14,52 +16,24 @@ import (
"Open_IM/pkg/common/tokenverify"
"Open_IM/pkg/common/tracelog"
pbFriend "Open_IM/pkg/proto/friend"
pbUser "Open_IM/pkg/proto/user"
"Open_IM/pkg/utils"
"context"
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"net"
"strconv"
"strings"
"Open_IM/internal/common/check"
"github.com/OpenIMSDK/getcdv3"
"google.golang.org/grpc"
)
type friendServer struct {
rpcPort int
rpcRegisterName string
etcdSchema string
etcdAddr []string
*rpc_server.RpcServer
controller.FriendInterface
controller.BlackInterface
userRpc pbUser.UserClient
}
func NewFriendServer(port int) *friendServer {
log.NewPrivateLog(constant.LogFileName)
f := friendServer{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImFriendName,
etcdSchema: config.Config.Etcd.EtcdSchema,
etcdAddr: config.Config.Etcd.EtcdAddr,
}
ttl := 10
etcdClient, err := getcdv3.NewEtcdConn(config.Config.Etcd.EtcdSchema, strings.Join(f.etcdAddr, ","), config.Config.RpcRegisterIP, config.Config.Etcd.UserName, config.Config.Etcd.Password, port, ttl)
r, err := rpc_server.NewRpcServer(config.Config.RpcRegisterIP, port, config.Config.RpcRegisterName.OpenImFriendName, config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema)
if err != nil {
panic("NewEtcdConn failed" + err.Error())
panic(err)
}
err = etcdClient.RegisterEtcd("", f.rpcRegisterName)
if err != nil {
panic("NewEtcdConn failed" + err.Error())
}
etcdClient.SetDefaultEtcdConfig(config.Config.RpcRegisterName.OpenImUserName, config.Config.RpcPort.OpenImUserPort)
conn := etcdClient.GetConn("", config.Config.RpcRegisterName.OpenImUserName)
f.userRpc = pbUser.NewUserClient(conn)
//mysql init
var mysql relation.Mysql
var model relation.FriendGorm
@@ -71,7 +45,6 @@ func NewFriendServer(port int) *friendServer {
if err != nil {
panic("db init err:" + err.Error())
}
err = mysql.InitConn().AutoMigrateModel(&relationTb.BlackModel{})
if err != nil {
panic("db init err:" + err.Error())
@@ -81,28 +54,22 @@ func NewFriendServer(port int) *friendServer {
} else {
panic("db init err:" + "conn is nil")
}
f.FriendInterface = controller.NewFriendController(model.DB)
f.BlackInterface = controller.NewBlackController(model.DB)
return &f
return &friendServer{
RpcServer: r,
FriendInterface: controller.NewFriendController(model.DB),
BlackInterface: controller.NewBlackController(model.DB),
}
}
func (s *friendServer) Run() {
log.NewInfo("0", "friendServer run...")
listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network
listener, err := net.Listen("tcp", address)
operationID := utils.OperationIDGenerator()
log.NewInfo(operationID, "friendServer run...")
listener, address, err := rpc_server.GetTcpListen(config.Config.ListenIP, s.Port)
if err != nil {
panic("listening err:" + err.Error() + s.rpcRegisterName)
panic(err)
}
log.NewInfo("0", "listen ok ", address)
log.NewInfo(operationID, "listen ok ", address)
defer listener.Close()
//grpc server
var grpcOpts []grpc.ServerOption
@@ -122,7 +89,7 @@ func (s *friendServer) Run() {
pbFriend.RegisterFriendServer(srv, s)
err = srv.Serve(listener)
if err != nil {
log.NewError("0", "Serve failed ", err.Error(), listener)
log.NewError(operationID, "Serve failed ", err.Error(), listener)
return
}
}
@@ -240,7 +207,7 @@ func (s *friendServer) SetFriendRemark(ctx context.Context, req *pbFriend.SetFri
}
// ok
func (s *friendServer) GetDesignatedFriendsReq(ctx context.Context, req *pbFriend.GetDesignatedFriendsReq) (resp *pbFriend.GetDesignatedFriendsResp, err error) {
func (s *friendServer) GetDesignatedFriends(ctx context.Context, req *pbFriend.GetDesignatedFriendsReq) (resp *pbFriend.GetDesignatedFriendsResp, err error) {
resp = &pbFriend.GetDesignatedFriendsResp{}
if err := check.Access(ctx, req.UserID); err != nil {
return nil, err
+7 -13
View File
@@ -21,7 +21,6 @@ import (
"Open_IM/pkg/utils"
"context"
"fmt"
"github.com/OpenIMSDK/getcdv3"
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"math/big"
"net"
@@ -67,7 +66,7 @@ func NewGroupServer(port int) *groupServer {
if err != nil {
panic(err.Error())
}
registerIP, err := network.GetRpcIP(config.Config.RpcRegisterIP)
registerIP, err := network.GetRpcRegisterIP(config.Config.RpcRegisterIP)
g.registerCenter = zkClient
err = g.registerCenter.Register(config.Config.RpcRegisterName.OpenImGroupName, registerIP, port)
if err != nil {
@@ -81,20 +80,15 @@ func NewGroupServer(port int) *groupServer {
}
func (s *groupServer) Run() {
log.NewInfo("", "group rpc start ")
listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
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("", "listen network success, ", address, listener)
log.NewInfo(operationID, "listen network success, ", address, listener)
defer listener.Close()
//grpc server
@@ -121,10 +115,10 @@ func (s *groupServer) Run() {
pbGroup.RegisterGroupServer(srv, s)
err = srv.Serve(listener)
if err != nil {
log.NewError("", "Serve failed ", err.Error())
log.NewError(operationID, "Serve failed ", err.Error())
return
}
log.NewInfo("", "group rpc success")
log.NewInfo(operationID, "group rpc success")
}
func (s *groupServer) CheckGroupAdmin(ctx context.Context, groupID string) error {
+15 -44
View File
@@ -2,6 +2,7 @@ package user
import (
"Open_IM/internal/common/convert"
"Open_IM/internal/common/rpc_server"
chat "Open_IM/internal/rpc/msg"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
@@ -16,10 +17,6 @@ import (
pbUser "Open_IM/pkg/proto/user"
"Open_IM/pkg/utils"
"context"
"net"
"strconv"
"strings"
grpcPrometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"google.golang.org/grpc"
@@ -28,23 +25,19 @@ import (
type userServer struct {
rpcPort int
rpcRegisterName string
etcdSchema string
etcdAddr []string
*rpc_server.RpcServer
controller.UserInterface
}
func NewUserServer(port int) *userServer {
log.NewPrivateLog(constant.LogFileName)
u := userServer{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImUserName,
etcdSchema: config.Config.Etcd.EtcdSchema,
etcdAddr: config.Config.Etcd.EtcdAddr,
r, err := rpc_server.NewRpcServer(config.Config.RpcRegisterIP, port, config.Config.RpcRegisterName.OpenImUserName, config.Config.Zookeeper.ZkAddr, config.Config.Zookeeper.Schema)
if err != nil {
panic(err)
}
//mysql init
var mysql relation.Mysql
var model relation.UserGorm
err := mysql.InitConn().AutoMigrateModel(&model)
err = mysql.InitConn().AutoMigrateModel(&model)
if err != nil {
panic("db init err:" + err.Error())
}
@@ -53,27 +46,17 @@ func NewUserServer(port int) *userServer {
} else {
panic("db init err:" + "conn is nil")
}
u.UserInterface = controller.NewUserController(model.DB)
return &u
return &userServer{RpcServer: r, UserInterface: controller.NewUserController(model.DB)}
}
func (s *userServer) Run() {
log.NewInfo("", "rpc user start...")
listenIP := ""
if config.Config.ListenIP == "" {
listenIP = "0.0.0.0"
} else {
listenIP = config.Config.ListenIP
}
address := listenIP + ":" + strconv.Itoa(s.rpcPort)
//listener network
listener, err := net.Listen("tcp", address)
operationID := utils.OperationIDGenerator()
log.NewInfo(operationID, "rpc user start...")
listener, address, err := rpc_server.GetTcpListen(config.Config.ListenIP, s.Port)
if err != nil {
panic("listening err:" + err.Error() + s.rpcRegisterName)
panic(err)
}
log.NewInfo("", "listen network success, address ", address, listener)
log.NewInfo(operationID, "listen ok ", address)
defer listener.Close()
//grpc server
var grpcOpts []grpc.ServerOption
@@ -91,24 +74,12 @@ func (s *userServer) Run() {
defer srv.GracefulStop()
//Service registers with etcd
pbUser.RegisterUserServer(srv, s)
rpcRegisterIP := config.Config.RpcRegisterIP
if config.Config.RpcRegisterIP == "" {
rpcRegisterIP, err = utils.GetLocalIP()
if err != nil {
log.Error("", "GetLocalIP failed ", err.Error())
}
}
err = rpc.RegisterEtcd(s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName, 10, "")
if err != nil {
log.NewError("", "RegisterEtcd failed ", err.Error(), s.etcdSchema, strings.Join(s.etcdAddr, ","), rpcRegisterIP, s.rpcPort, s.rpcRegisterName)
panic(utils.Wrap(err, "register user module rpc to etcd err"))
}
err = srv.Serve(listener)
if err != nil {
log.NewError("", "Serve failed ", err.Error())
return
panic(err)
}
log.NewInfo("", "rpc user success")
log.NewInfo(operationID, "rpc user success")
}
// ok