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

internal/api/manage/management_user.go

@@ -17,9 +17,10 @@ import (
 	rpc "Open_IM/pkg/proto/user"
 	"Open_IM/pkg/utils"
 	"context"
-	"github.com/gin-gonic/gin"
 	"net/http"
 	"strings"
+
+	"github.com/gin-gonic/gin"
 )
 
 func DeleteUser(c *gin.Context) {
@@ -198,7 +199,7 @@ func GetUsersOnlineStatus(c *gin.Context) {
 	var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
 	var respResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
 	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 {
 		client := pbRelay.NewRelayClient(v)
 		reply, err := client.GetUsersOnlineStatus(context.Background(), req)

internal/api/user/user.go

@@ -14,9 +14,10 @@ import (
 	rpc "Open_IM/pkg/proto/user"
 	"Open_IM/pkg/utils"
 	"context"
-	"github.com/gin-gonic/gin"
 	"net/http"
 	"strings"
+
+	"github.com/gin-gonic/gin"
 )
 
 func GetUsersInfoFromCache(c *gin.Context) {
@@ -415,7 +416,7 @@ func GetUsersOnlineStatus(c *gin.Context) {
 	var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
 	var respResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
 	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 {
 		log.Debug(params.OperationID, "get node ", *v, v.Target())
 		client := pbRelay.NewRelayClient(v)

internal/msg_gateway/gate/ws_server.go

@@ -12,9 +12,10 @@ import (
 	"bytes"
 	"context"
 	"encoding/gob"
+	"strings"
+
 	go_redis "github.com/go-redis/redis/v8"
 	"github.com/pkg/errors"
-	"strings"
 
 	//"gopkg.in/errgo.v2/errors"
 	"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) {
-	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)
 	for _, v := range grpcCons {
 		if v.Target() == rpcSvr.target {

internal/push/logic/push_to_client.go

@@ -43,7 +43,7 @@ func MsgToUser(pushMsg *pbPush.PushMsgReq) {
 	var wsResult []*pbRelay.SingelMsgToUserResultList
 	isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush)
 	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}
 	callbackResp := callbackOnlinePush(pushMsg.OperationID, UIDList, pushMsg.MsgData)
@@ -188,7 +188,7 @@ func MsgToSuperGroupUser(pushMsg *pbPush.PushMsgReq) {
 		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
 	log.Debug(pushMsg.OperationID, "len  grpc", len(grpcCons), "data", pushMsg.String())

internal/rpc/auth/auth.go

@@ -90,7 +90,7 @@ func (rpc *rpcAuth) ForceLogout(_ context.Context, req *pbAuth.ForceLogoutReq) (
 
 func (rpc *rpcAuth) forceKickOff(userID string, platformID int32, operationID string) error {
 	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 {
 		client := pbRelay.NewRelayClient(v)
 		kickReq := &pbRelay.KickUserOfflineReq{OperationID: operationID, KickUserIDList: []string{userID}, PlatformID: platformID}

internal/rpc/msg/send_msg.go

@@ -17,13 +17,14 @@ import (
 	"Open_IM/pkg/utils"
 	"context"
 	"errors"
-	go_redis "github.com/go-redis/redis/v8"
-	"github.com/golang/protobuf/proto"
 	"math/rand"
 	"strconv"
 	"strings"
 	"sync"
 	"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
@@ -884,7 +885,7 @@ func getOnlineAndOfflineUserIDList(memberList []string, m map[string][]string, o
 	req.OperationID = operationID
 	req.OpUserID = config.Config.Manager.AppManagerUid[0]
 	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 {
 		client := pbRelay.NewRelayClient(v)
 		reply, err := client.GetUsersOnlineStatus(context.Background(), req)

pkg/common/db/RedisModel.go

@@ -11,11 +11,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
+	"time"
+
 	go_redis "github.com/go-redis/redis/v8"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/golang/protobuf/proto"
-	"strconv"
-	"time"
 )
 
 const (
@@ -230,16 +231,14 @@ func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string,
 }
 func (d *DataBases) DeleteMessageFromCache(msgList []*pbChat.MsgDataToMQ, uid string, operationID string) error {
 	ctx := context.Background()
-	var keys []string
 	for _, msg := range msgList {
 		key := messageCache + uid + "_" + strconv.Itoa(int(msg.MsgData.Seq))
-		keys = append(keys, key)
+		err := d.RDB.Del(ctx, key).Err()
-	}
-	err := d.RDB.Del(ctx, keys...).Err()
 		if err != nil {
-		log2.NewWarn(operationID, utils.GetSelfFuncName(), "redis failed", "args:", keys, uid, err.Error(), msgList)
+			log2.NewWarn(operationID, utils.GetSelfFuncName(), "redis failed", "args:", key, uid, err.Error(), msgList)
 		}
-	return err
+	}
+	return nil
 }
 
 func (d *DataBases) CleanUpOneUserAllMsgFromRedis(userID string, operationID string) error {

pkg/grpc-etcdv3/getcdv3/resolver.go

@@ -6,17 +6,19 @@ import (
 	"Open_IM/pkg/utils"
 	"context"
 	"fmt"
+
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 
 	//"go.etcd.io/etcd/mvcc/mvccpb"
 	//"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"
 	"sync"
 	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/balancer/roundrobin"
+	"google.golang.org/grpc/resolver"
 )
 
 type Resolver struct {
@@ -176,6 +178,7 @@ func GetConfigConn(serviceName string, operationID string) *grpc.ClientConn {
 		log.Error(operationID, "grpc.Dail failed ", err.Error())
 		return nil
 	}
+	log.NewDebug(operationID, utils.GetSelfFuncName(), serviceName, conn)
 	return conn
 }
 
@@ -274,17 +277,18 @@ 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 {
 		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)
+	log.NewDebug(operationID, utils.GetSelfFuncName(), config.Config.RpcRegisterName.OpenImRelayName, grpcConns)
 	return grpcConns
 }
 
-func getConn4UniqueFromConfig(servicename, operationID string) []*grpc.ClientConn {
+func GetDefaultGatewayConn4UniqueFromcfg(operationID string) []*grpc.ClientConn {
 	rpcRegisterIP := config.Config.RpcRegisterIP
 	var err error
 	if config.Config.RpcRegisterIP == "" {
@@ -294,7 +298,21 @@ func getConn4UniqueFromConfig(servicename, operationID string) []*grpc.ClientCon
 			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 {