blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

etcd

Browse Source
This commit is contained in:
wangchuxiao 2022-08-17 14:57:57 +08:00
parent 9138f90843
commit d0a200f652
7 changed files with 41 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/api/manage/management_user.go
View File

@ -17,9 +17,10 @@ import (
rpc "Open_IM/pkg/proto/user" rpc "Open_IM/pkg/proto/user"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
"github.com/gin-gonic/gin"
"net/http" "net/http"
"strings" "strings"
"github.com/gin-gonic/gin"
) )
func DeleteUser(c *gin.Context) { func DeleteUser(c *gin.Context) {
@ -198,7 +199,7 @@ func GetUsersOnlineStatus(c *gin.Context) {
var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
var respResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult var respResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
flag := false flag := false
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImRelayName) grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), params.OperationID)
for _, v := range grpcCons { for _, v := range grpcCons {
client := pbRelay.NewRelayClient(v) client := pbRelay.NewRelayClient(v)
reply, err := client.GetUsersOnlineStatus(context.Background(), req) reply, err := client.GetUsersOnlineStatus(context.Background(), req)

5
internal/api/user/user.go
View File

@ -14,9 +14,10 @@ import (
rpc "Open_IM/pkg/proto/user" rpc "Open_IM/pkg/proto/user"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
"github.com/gin-gonic/gin"
"net/http" "net/http"
"strings" "strings"
"github.com/gin-gonic/gin"
) )
func GetUsersInfoFromCache(c *gin.Context) { func GetUsersInfoFromCache(c *gin.Context) {
@ -415,7 +416,7 @@ func GetUsersOnlineStatus(c *gin.Context) {
var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
var respResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult var respResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
flag := false flag := false
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImRelayName) grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), params.OperationID)
for _, v := range grpcCons { for _, v := range grpcCons {
log.Debug(params.OperationID, "get node ", *v, v.Target()) log.Debug(params.OperationID, "get node ", *v, v.Target())
client := pbRelay.NewRelayClient(v) client := pbRelay.NewRelayClient(v)

5
internal/msg_gateway/gate/ws_server.go
View File

@ -12,9 +12,10 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/gob" "encoding/gob"
"strings"
go_redis "github.com/go-redis/redis/v8" go_redis "github.com/go-redis/redis/v8"
"github.com/pkg/errors" "github.com/pkg/errors"
"strings"
//"gopkg.in/errgo.v2/errors" //"gopkg.in/errgo.v2/errors"
"net/http" "net/http"
@ -119,7 +120,7 @@ func (ws *WServer) SetWriteTimeoutWriteMsg(conn *UserConn, a int, msg []byte, ti
} }
func (ws *WServer) MultiTerminalLoginRemoteChecker(userID string, platformID int32, token string, operationID string) { func (ws *WServer) MultiTerminalLoginRemoteChecker(userID string, platformID int32, token string, operationID string) {
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImRelayName) grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), operationID)
log.NewInfo(operationID, utils.GetSelfFuncName(), "args grpcCons: ", userID, platformID, grpcCons) log.NewInfo(operationID, utils.GetSelfFuncName(), "args grpcCons: ", userID, platformID, grpcCons)
for _, v := range grpcCons { for _, v := range grpcCons {
if v.Target() == rpcSvr.target { if v.Target() == rpcSvr.target {

4
internal/push/logic/push_to_client.go
View File

@ -43,7 +43,7 @@ func MsgToUser(pushMsg *pbPush.PushMsgReq) {
var wsResult []*pbRelay.SingelMsgToUserResultList var wsResult []*pbRelay.SingelMsgToUserResultList
isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush) isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush)
log.Debug(pushMsg.OperationID, "Get msg from msg_transfer And push msg", pushMsg.String()) log.Debug(pushMsg.OperationID, "Get msg from msg_transfer And push msg", pushMsg.String())
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImRelayName) grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), pushMsg.OperationID)
var UIDList = []string{pushMsg.PushToUserID} var UIDList = []string{pushMsg.PushToUserID}
callbackResp := callbackOnlinePush(pushMsg.OperationID, UIDList, pushMsg.MsgData) callbackResp := callbackOnlinePush(pushMsg.OperationID, UIDList, pushMsg.MsgData)
@ -188,7 +188,7 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
pushToUserIDList = cacheResp.UserIDList pushToUserIDList = cacheResp.UserIDList
} }
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImRelayName) grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), pushMsg.OperationID)
//Online push message //Online push message
log.Debug(pushMsg.OperationID, "len grpc", len(grpcCons), "data", pushMsg.String()) log.Debug(pushMsg.OperationID, "len grpc", len(grpcCons), "data", pushMsg.String())

2
internal/rpc/auth/auth.go
View File

@ -90,7 +90,7 @@ func (rpc *rpcAuth) ForceLogout(_ context.Context, req *pbAuth.ForceLogoutReq) (
func (rpc *rpcAuth) forceKickOff(userID string, platformID int32, operationID string) error { func (rpc *rpcAuth) forceKickOff(userID string, platformID int32, operationID string) error {
log.NewInfo(operationID, utils.GetSelfFuncName(), " args ", userID, platformID) log.NewInfo(operationID, utils.GetSelfFuncName(), " args ", userID, platformID)
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImRelayName) grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), operationID)
for _, v := range grpcCons { for _, v := range grpcCons {
client := pbRelay.NewRelayClient(v) client := pbRelay.NewRelayClient(v)
kickReq := &pbRelay.KickUserOfflineReq{OperationID: operationID, KickUserIDList: []string{userID}, PlatformID: platformID} kickReq := &pbRelay.KickUserOfflineReq{OperationID: operationID, KickUserIDList: []string{userID}, PlatformID: platformID}

7
internal/rpc/msg/send_msg.go
View File

@ -17,13 +17,14 @@ import (
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
"errors" "errors"
go_redis "github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto"
"math/rand" "math/rand"
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
"time" "time"
go_redis "github.com/go-redis/redis/v8"
"github.com/golang/protobuf/proto"
) )
//When the number of group members is greater than this value，Online users will be sent first，Guaranteed service availability //When the number of group members is greater than this value，Online users will be sent first，Guaranteed service availability
@ -884,7 +885,7 @@ func getOnlineAndOfflineUserIDList(memberList []string, m map[string][]string, o
req.OperationID = operationID req.OperationID = operationID
req.OpUserID = config.Config.Manager.AppManagerUid[0] req.OpUserID = config.Config.Manager.AppManagerUid[0]
flag := false flag := false
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImRelayName) grpcCons := getcdv3.GetDefaultGatewayConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), operationID)
for _, v := range grpcCons { for _, v := range grpcCons {
client := pbRelay.NewRelayClient(v) client := pbRelay.NewRelayClient(v)
reply, err := client.GetUsersOnlineStatus(context.Background(), req) reply, err := client.GetUsersOnlineStatus(context.Background(), req)

34
pkg/grpc-etcdv3/getcdv3/resolver.go
View File

@ -6,17 +6,19 @@ import (
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"context" "context"
"fmt" "fmt"
"go.etcd.io/etcd/api/v3/mvccpb" "go.etcd.io/etcd/api/v3/mvccpb"
clientv3 "go.etcd.io/etcd/client/v3" clientv3 "go.etcd.io/etcd/client/v3"
//"go.etcd.io/etcd/mvcc/mvccpb" //"go.etcd.io/etcd/mvcc/mvccpb"
//"google.golang.org/genproto/googleapis/ads/googleads/v1/services" //"google.golang.org/genproto/googleapis/ads/googleads/v1/services"
"google.golang.org/grpc"
"google.golang.org/grpc/balancer/roundrobin"
"google.golang.org/grpc/resolver"
"strings" "strings"
"sync" "sync"
"time" "time"
"google.golang.org/grpc"
"google.golang.org/grpc/balancer/roundrobin"
"google.golang.org/grpc/resolver"
) )
type Resolver struct { type Resolver struct {
@ -274,17 +276,17 @@ func (r *Resolver) watch(prefix string, addrList []resolver.Address) {
} }
} }
func GetDefaultConn4Unique(schema, etcdaddr, servicename, operationID string) []*grpc.ClientConn { func GetDefaultGatewayConn4Unique(schema, etcdaddr, operationID string) []*grpc.ClientConn {
grpcConns := getConn4Unique(schema, etcdaddr, servicename) grpcConns := getConn4Unique(schema, etcdaddr, config.Config.RpcRegisterName.OpenImRelayName)
if len(grpcConns) > 0 { if len(grpcConns) > 0 {
return grpcConns return grpcConns
} }
log.NewWarn(operationID, utils.GetSelfFuncName(), " len(grpcConns) < 0 ", schema, etcdaddr, servicename) log.NewWarn(operationID, utils.GetSelfFuncName(), " len(grpcConns) < 0 ", schema, etcdaddr, config.Config.RpcRegisterName.OpenImRelayName)
grpcConns = getConn4UniqueFromConfig(servicename, operationID) grpcConns = GetDefaultGatewayConn4UniqueFromcfg(operationID)
return grpcConns return grpcConns
} }
func getConn4UniqueFromConfig(servicename, operationID string) []*grpc.ClientConn { func GetDefaultGatewayConn4UniqueFromcfg(operationID string) []*grpc.ClientConn {
rpcRegisterIP := config.Config.RpcRegisterIP rpcRegisterIP := config.Config.RpcRegisterIP
var err error var err error
if config.Config.RpcRegisterIP == "" { if config.Config.RpcRegisterIP == "" {
@ -294,7 +296,21 @@ func getConn4UniqueFromConfig(servicename, operationID string) []*grpc.ClientCon
return nil return nil
} }
} }
return nil var conns []*grpc.ClientConn
configPortList := config.Config.RpcPort.OpenImMessageGatewayPort
for _, port := range configPortList {
target := rpcRegisterIP + ":" + utils.Int32ToString(int32(port))
log.Info(operationID, "rpcRegisterIP ", rpcRegisterIP, " port ", configPortList, " grpc target: ", target, " serviceName: ", "msgGateway")
conn, err := grpc.Dial(target, grpc.WithInsecure())
if err != nil {
log.Error(operationID, "grpc.Dail failed ", err.Error())
continue
}
conns = append(conns, conn)
}
return conns
} }
func getConn4Unique(schema, etcdaddr, servicename string) []*grpc.ClientConn { func getConn4Unique(schema, etcdaddr, servicename string) []*grpc.ClientConn {