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

pkg/common/db/newRedisModel.go

@@ -155,6 +155,7 @@ func (d *DataBases) GetMessageListBySeq(userID string, seqList []uint32, operati
 }
 func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string, operationID string) error {
 	ctx := context.Background()
+	pipe := d.rdb.Pipeline()
 	var failedList []pbChat.MsgDataToMQ
 	for _, msg := range msgList {
 		key := messageCache + uid + "_" + strconv.Itoa(int(msg.MsgData.Seq))
@@ -164,7 +165,7 @@ func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string,
 			continue
 		}
 		log2.NewDebug(operationID, "convert string is ", s)
-		err = d.rdb.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err()
+		err = pipe.Set(ctx, key, s, time.Duration(config.Config.MsgCacheTimeout)*time.Second).Err()
 		//err = d.rdb.HMSet(context.Background(), "12", map[string]interface{}{"1": 2, "343": false}).Err()
 		if err != nil {
 			log2.NewWarn(operationID, utils.GetSelfFuncName(), "redis failed", "args:", key, *msg, uid, s, err.Error())
@@ -174,7 +175,8 @@ func (d *DataBases) SetMessageToCache(msgList []*pbChat.MsgDataToMQ, uid string,
 	if len(failedList) != 0 {
 		return errors.New(fmt.Sprintf("set msg to cache failed, failed lists: %q,%s", failedList, operationID))
 	}
-	return nil
+	_, err := pipe.Exec(ctx)
+	return err
 }
 
 func (d *DataBases) CleanUpOneUserAllMsgFromRedis(userID string, operationID string) error {

pkg/common/kafka/producer.go

@@ -2,6 +2,7 @@ package kafka
 
 import (
 	log2 "Open_IM/pkg/common/log"
+	"Open_IM/pkg/utils"
 	"errors"
 	"github.com/Shopify/sarama"
 	"github.com/golang/protobuf/proto"
@@ -16,8 +17,9 @@ type Producer struct {
 
 func NewKafkaProducer(addr []string, topic string) *Producer {
 	p := Producer{}
-	p.config = sarama.NewConfig()                             //Instantiate a sarama Config
-	p.config.Producer.Return.Successes = true                 //Whether to enable the successes channel to be notified after the message is sent successfully
+	p.config = sarama.NewConfig()             //Instantiate a sarama Config
+	p.config.Producer.Return.Successes = true //Whether to enable the successes channel to be notified after the message is sent successfully
+	p.config.Producer.Return.Errors = true
 	p.config.Producer.RequiredAcks = sarama.WaitForAll        //Set producer Message Reply level 0 1 all
 	p.config.Producer.Partitioner = sarama.NewHashPartitioner //Set the hash-key automatic hash partition. When sending a message, you must specify the key value of the message. If there is no key, the partition will be selected randomly
 
@@ -44,11 +46,16 @@ func (p *Producer) SendMessage(m proto.Message, key string, operationID string)
 		return -1, -1, err
 	}
 	if len(bMsg) == 0 {
-		return 0, 0, errors.New("msg content is nil")
+		log2.Error(operationID, "len(bMsg) == 0 ")
+		return 0, 0, errors.New("len(bMsg) == 0 ")
 	}
 	kMsg.Value = sarama.ByteEncoder(bMsg)
-	log2.Info(operationID, "ByteEncoder SendMessage begin", "key ", kMsg, p.producer)
+	log2.Info(operationID, "ByteEncoder SendMessage begin", "key ", kMsg, p.producer, "len: ", kMsg.Key.Length(), kMsg.Value.Length())
+	if kMsg.Key.Length() == 0 || kMsg.Value.Length() == 0 {
+		log2.Error(operationID, "kMsg.Key.Length() == 0 || kMsg.Value.Length() == 0 ", kMsg)
+		return -1, -1, errors.New("key or value == 0")
+	}
 	a, b, c := p.producer.SendMessage(kMsg)
-	log2.Info(operationID, "ByteEncoder SendMessage end", "key ", kMsg, p.producer)
-	return a, b, c
+	log2.Info(operationID, "ByteEncoder SendMessage end", "key ", kMsg.Key.Length(), kMsg.Value.Length(), p.producer)
+	return a, b, utils.Wrap(c, "")
 }

pkg/common/token_verify/jwt_token.go

@@ -245,15 +245,19 @@ func VerifyToken(token, uid string) (bool, error) {
 	return true, nil
 }
 func WsVerifyToken(token, uid string, platformID string, operationID string) (bool, error, string) {
+	argMsg := "token: " + token + " operationID: " + operationID + " userID: " + uid + " platformID: " + platformID
 	claims, err := ParseToken(token, operationID)
 	if err != nil {
-		return false, utils.Wrap(err, "parse token err"), "parse token err"
+		errMsg := "parse token err " + argMsg
+		return false, utils.Wrap(err, errMsg), errMsg
 	}
 	if claims.UID != uid {
-		return false, utils.Wrap(&constant.ErrTokenUnknown, "uid is not same to token uid"), "uid is not same to token uid"
+		errMsg := " uid is not same to token uid " + " claims.UID " + claims.UID + argMsg
+		return false, utils.Wrap(&constant.ErrTokenUnknown, errMsg), errMsg
 	}
 	if claims.Platform != constant.PlatformIDToName(utils.StringToInt(platformID)) {
-		return false, utils.Wrap(&constant.ErrTokenUnknown, "platform is not same to token platform"), "platform is not same to token platform"
+		errMsg := " platform is not same to token platform " + argMsg + "claims platformID " + claims.Platform
+		return false, utils.Wrap(&constant.ErrTokenUnknown, errMsg), errMsg
 	}
 	log.NewDebug(operationID, utils.GetSelfFuncName(), " check ok ", claims.UID, uid, claims.Platform)
 	return true, nil, ""