Group send message failed when messagecallback is enable,

[zengzhengrong]

#token config
tokenpolicy:
  accessSecret: "open_im_server"
  # Token effective time seconds as a unit
  #Seven days   7*24*60*60
  accessExpire: 604800

messagecallback:
  callbackSwitch: true
  callbackUrl: "xxx/callback/"

source code link

	if config.Config.MessageCallBack.CallbackSwitch {
		bMsg, err := http2.Post(config.Config.MessageCallBack.CallbackUrl, MsgCallBackReq{
			SendID:      pb.SendID,
			RecvID:      pb.RecvID,
			Content:     pb.Content,
			SendTime:    pbData.SendTime,
			MsgFrom:     pbData.MsgFrom,
			ContentType: pb.ContentType,
			SessionType: pb.SessionType,
			PlatformID:  pb.PlatformID,
		}, "application/json; charset=utf-8")
		if err != nil {
			log.ErrorByKv("callback to Business server err", pb.OperationID, "args", pb.String(), "err", err.Error())
			return returnMsg(&replay, pb, http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), "", 0)
		} else if err = json.Unmarshal(bMsg, &m); err != nil {
			log.ErrorByKv("ws json Unmarshal err", pb.OperationID, "args", pb.String(), "err", err.Error())
			return returnMsg(&replay, pb, 200, err.Error(), "", 0)
		} else {
			if m.ErrCode != 0 {
				return returnMsg(&replay, pb, m.ResponseErrCode, m.ErrMsg, "", 0)
			} else {
				pbData.Content = m.ResponseResult.ModifiedMsg
				err1 := rpc.sendMsgToKafka(&pbData, pbData.RecvID)
				err2 := rpc.sendMsgToKafka(&pbData, pbData.SendID)
				if err1 != nil || err2 != nil {
					return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
				}
				return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
			}
		}
	} else {
		switch pbData.SessionType {
		case constant.SingleChatType:
			err1 := rpc.sendMsgToKafka(&pbData, pbData.RecvID)
			err2 := rpc.sendMsgToKafka(&pbData, pbData.SendID)
			if err1 != nil || err2 != nil {
				return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
			}
			return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
		case constant.GroupChatType:
			etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
			client := pbGroup.NewGroupClient(etcdConn)
			req := &pbGroup.GetGroupAllMemberReq{
				GroupID:     pbData.RecvID,
				Token:       pbData.Token,
				OperationID: pbData.OperationID,
			}
			reply, err := client.GetGroupAllMember(context.Background(), req)
			if err != nil {
				log.Error(pbData.Token, pbData.OperationID, "rpc send_msg getGroupInfo failed, err = %s", err.Error())
				return returnMsg(&replay, pb, 201, err.Error(), "", 0)
			}
			if reply.ErrorCode != 0 {
				log.Error(pbData.Token, pbData.OperationID, "rpc send_msg getGroupInfo failed, err = %s", reply.ErrorMsg)
				return returnMsg(&replay, pb, reply.ErrorCode, reply.ErrorMsg, "", 0)
			}
			var addUidList []string
			switch pbData.ContentType {
			case constant.KickGroupMemberTip:
				var notification content_struct.NotificationContent
				var kickContent group.KickGroupMemberReq
				err := utils.JsonStringToStruct(pbData.Content, &notification)
				if err != nil {
					log.ErrorByKv("json unmarshall err", pbData.OperationID, "err", err.Error())
					return returnMsg(&replay, pb, 200, err.Error(), "", 0)
				} else {
					err := utils.JsonStringToStruct(notification.Detail, &kickContent)
					if err != nil {
						log.ErrorByKv("json unmarshall err", pbData.OperationID, "err", err.Error())
						return returnMsg(&replay, pb, 200, err.Error(), "", 0)
					}
					for _, v := range kickContent.UidListInfo {
						addUidList = append(addUidList, v.UserId)
					}
				}
			case constant.QuitGroupTip:
				addUidList = append(addUidList, pbData.SendID)
			default:
			}
			groupID := pbData.RecvID
			for i, v := range reply.MemberList {
				pbData.RecvID = v.UserId + " " + groupID
				err := rpc.sendMsgToKafka(&pbData, utils.IntToString(i))
				if err != nil {
					return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
				}
			}
			for i, v := range addUidList {
				pbData.RecvID = v + " " + groupID
				err := rpc.sendMsgToKafka(&pbData, utils.IntToString(i+1))
				if err != nil {
					return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
				}
			}
			return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
		default:

		}

	}

if callbackSwitch: true , It does not send meesage to group

[zengzhengrong]

#18