management modify

internal/api/manage/management_user.go

@@ -6,154 +6,175 @@
  */
 package manage
 
-import "github.com/gin-gonic/gin"
+import (
+	api "Open_IM/pkg/base_info"
+	"Open_IM/pkg/common/config"
+	"Open_IM/pkg/common/constant"
+	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/token_verify"
+	"Open_IM/pkg/grpc-etcdv3/getcdv3"
+	pbRelay "Open_IM/pkg/proto/relay"
+	rpc "Open_IM/pkg/proto/user"
+	"Open_IM/pkg/utils"
+	"context"
+	"github.com/gin-gonic/gin"
+	"net/http"
+	"strings"
+)
 
 func DeleteUser(c *gin.Context) {
+	params := api.DeleteUsersReq{}
+	if err := c.BindJSON(&params); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	req := &rpc.DeleteUsersReq{}
+	utils.CopyStructFields(req, &params)
+	var ok bool
+	ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"))
+	if !ok {
+		log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
+		return
+	}
+	log.NewInfo(params.OperationID, "DeleteUser args ", req.String())
+	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
+	client := rpc.NewUserClient(etcdConn)
+
+	RpcResp, err := client.DeleteUsers(context.Background(), req)
+	if err != nil {
+		log.NewError(req.OperationID, "call delete users rpc server failed", err.Error())
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call delete users rpc server failed"})
+		return
+	}
+	failedUserIDList := make([]string, 0)
+	for _, v := range RpcResp.FailedUserIDList {
+		failedUserIDList = append(failedUserIDList, v)
+	}
+	resp := api.DeleteUsersResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}}
+	resp.FailedUserIDList = failedUserIDList
+	log.NewInfo(req.OperationID, "DeleteUser api return", resp)
+	c.JSON(http.StatusOK, resp)
+}
+func GetAllUsersUid(c *gin.Context) {
+	params := api.GetAllUsersUidReq{}
+	if err := c.BindJSON(&params); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	req := &rpc.GetAllUserIDReq{}
+	utils.CopyStructFields(req, &params)
+	var ok bool
+	ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"))
+	if !ok {
+		log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
+		return
+	}
+	log.NewInfo(params.OperationID, "GetAllUsersUid args ", req.String())
+	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
+	client := rpc.NewUserClient(etcdConn)
+	RpcResp, err := client.GetAllUserID(context.Background(), req)
+	if err != nil {
+		log.NewError(req.OperationID, "call GetAllUsersUid users rpc server failed", err.Error())
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call GetAllUsersUid users rpc server failed"})
+		return
+	}
+	userIDList := make([]string, 0)
+	for _, v := range RpcResp.UserIDList {
+		userIDList = append(userIDList, v)
+	}
+	resp := api.GetAllUsersUidResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}}
+	resp.UserIDList = userIDList
+	log.NewInfo(req.OperationID, "GetAllUsersUid api return", resp)
+	c.JSON(http.StatusOK, resp)
 
 }
+func AccountCheck(c *gin.Context) {
+	params := api.AccountCheckReq{}
+	if err := c.BindJSON(&params); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	req := &rpc.AccountCheckReq{}
+	utils.CopyStructFields(req, &params)
+	var ok bool
+	ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"))
+	if !ok {
+		log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
+		return
+	}
+	log.NewInfo(params.OperationID, "AccountCheck args ", req.String())
+	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
+	client := rpc.NewUserClient(etcdConn)
 
-//
-//func DeleteUser(c *gin.Context) {
-//	params := paramsDeleteUsers{}
-//	if err := c.BindJSON(&params); err != nil {
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
-//		return
-//	}
-//	log.InfoByKv("DeleteUser req come here", params.OperationID, "DeleteUidList", params.DeleteUidList)
-//	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
-//	client := pbUser.NewUserClient(etcdConn)
-//	//defer etcdConn.Close()
-//
-//	req := &pbUser.DeleteUsersReq{
-//		OperationID:   params.OperationID,
-//		DeleteUidList: params.DeleteUidList,
-//		Token:         c.Request.Header.Get("token"),
-//	}
-//	RpcResp, err := client.DeleteUsers(context.Background(), req)
-//	if err != nil {
-//		log.NewError(req.OperationID, "call delete users rpc server failed", err.Error())
-//		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call delete users rpc server failed"})
-//		return
-//	}
-//	failedUidList := make([]string, 0)
-//	for _, v := range RpcResp.FailedUidList {
-//		failedUidList = append(failedUidList, v)
-//	}
-//	log.InfoByKv("call delete user rpc server is success", params.OperationID, "resp args", RpcResp.String())
-//	resp := gin.H{"errCode": RpcResp.CommonResp.ErrorCode, "errMsg": RpcResp.CommonResp.ErrorMsg, "failedUidList": RpcResp.FailedUidList}
-//	c.JSON(http.StatusOK, resp)
-//}
-//func GetAllUsersUid(c *gin.Context) {
-//	params := paramsGetAllUsersUid{}
-//	if err := c.BindJSON(&params); err != nil {
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
-//		return
-//	}
-//	log.InfoByKv("GetAllUsersUid req come here", params.OperationID)
-//	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
-//	client := pbUser.NewUserClient(etcdConn)
-//	//defer etcdConn.Close()
-//
-//	req := &pbUser.GetAllUsersUidReq{
-//		OperationID: params.OperationID,
-//		Token:       c.Request.Header.Get("token"),
-//	}
-//	RpcResp, err := client.GetAllUsersUid(context.Background(), req)
-//	if err != nil {
-//		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error(), "uidList": []string{}})
-//		return
-//	}
-//	log.InfoByKv("call GetAllUsersUid rpc server is success", params.OperationID, "resp args", RpcResp.String())
-//	resp := gin.H{"errCode": RpcResp.CommonResp.ErrorCode, "errMsg": RpcResp.CommonResp.ErrorMsg, "uidList": RpcResp.UidList}
-//	c.JSON(http.StatusOK, resp)
-//
-//}
-//func AccountCheck(c *gin.Context) {
-//	params := paramsAccountCheck{}
-//	if err := c.BindJSON(&params); err != nil {
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
-//		return
-//	}
-//	log.InfoByKv("AccountCheck req come here", params.OperationID, params.UserIDList)
-//	etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImUserName)
-//	client := pbUser.NewUserClient(etcdConn)
-//	//defer etcdConn.Close()
-//
-//	req := &pbUser.AccountCheckReq{
-//		OperationID: params.OperationID,
-//		Token:       c.Request.Header.Get("token"),
-//		UidList:     params.UserIDList,
-//	}
-//	RpcResp, err := client.AccountCheck(context.Background(), req)
-//	if err != nil {
-//		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
-//		return
-//	}
-//	log.InfoByKv("call AccountCheck rpc server is success", params.OperationID, "resp args", RpcResp.String())
-//	resp := gin.H{"errCode": RpcResp.CommonResp.ErrorCode, "errMsg": RpcResp.CommonResp.ErrorMsg, "result": RpcResp.Result}
-//	c.JSON(http.StatusOK, resp)
-//
-//}
-//func GetUsersOnlineStatus(c *gin.Context) {
-//	params := paramsGetUsersOnlineStatus{}
-//	if err := c.BindJSON(&params); err != nil {
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
-//		return
-//	}
-//	claims, err := token_verify.ParseToken(c.Request.Header.Get("token"))
-//	if err != nil {
-//		log.ErrorByKv("parse token failed", params.OperationID, "err", err.Error())
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 401, "errMsg": err.Error()})
-//		return
-//	}
-//	if !utils.IsContain(claims.UID, config.Config.Manager.AppManagerUid) {
-//		log.ErrorByKv(" Authentication failed", params.OperationID, "args", c)
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 402, "errMsg": "not authorized"})
-//		return
-//	}
-//	req := &pbRelay.GetUsersOnlineStatusReq{
-//		OperationID: params.OperationID,
-//		UserIDList:  params.UserIDList,
-//	}
-//	var wsResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
-//	var respResult []*pbRelay.GetUsersOnlineStatusResp_SuccessResult
-//	flag := false
-//	log.NewDebug(params.OperationID, "GetUsersOnlineStatus req come here", params.UserIDList)
-//	grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
-//	for _, v := range grpcCons {
-//		client := pbRelay.NewOnlineMessageRelayServiceClient(v)
-//		reply, err := client.GetUsersOnlineStatus(context.Background(), req)
-//		if err != nil {
-//			log.NewError(params.OperationID, "GetUsersOnlineStatus rpc  err", req.String(), err.Error())
-//			continue
-//		} else {
-//			if reply.ErrCode == 0 {
-//				wsResult = append(wsResult, reply.SuccessResult...)
-//			}
-//		}
-//	}
-//	log.NewDebug(params.OperationID, "call GetUsersOnlineStatus rpc server is success", wsResult)
-//	//Online data merge of each node
-//	for _, v1 := range params.UserIDList {
-//		flag = false
-//		temp := new(pbRelay.GetUsersOnlineStatusResp_SuccessResult)
-//		for _, v2 := range wsResult {
-//			if v2.UserID == v1 {
-//				flag = true
-//				temp.UserID = v1
-//				temp.Status = constant.OnlineStatus
-//				temp.DetailPlatformStatus = append(temp.DetailPlatformStatus, v2.DetailPlatformStatus...)
-//			}
-//
-//		}
-//		if !flag {
-//			temp.UserID = v1
-//			temp.Status = constant.OfflineStatus
-//		}
-//		respResult = append(respResult, temp)
-//	}
-//	log.NewDebug(params.OperationID, "Finished merged data", respResult)
-//	resp := gin.H{"errCode": 0, "errMsg": "", "successResult": respResult}
-//	c.JSON(http.StatusOK, resp)
-//
-//}
+	RpcResp, err := client.AccountCheck(context.Background(), req)
+	if err != nil {
+		log.NewError(req.OperationID, "call AccountCheck users rpc server failed", err.Error())
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "call AccountCheck users rpc server failed"})
+		return
+	}
+	resp := api.AccountCheckResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}}
+	resp.ResultList = RpcResp.ResultList
+	log.NewInfo(req.OperationID, "AccountCheck api return", resp)
+	c.JSON(http.StatusOK, resp)
+}
+func GetUsersOnlineStatus(c *gin.Context) {
+	params := api.GetUsersOnlineStatusReq{}
+	if err := c.BindJSON(&params); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
+		return
+	}
+	req := &pbRelay.GetUsersOnlineStatusReq{}
+	utils.CopyStructFields(req, &params)
+	var ok bool
+	ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"))
+	if !ok {
+		log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
+		c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
+		return
+	}
+	log.NewInfo(params.OperationID, "GetUsersOnlineStatus args ", req.String())
+	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.OpenImOnlineMessageRelayName)
+	for _, v := range grpcCons {
+		client := pbRelay.NewOnlineMessageRelayServiceClient(v)
+		reply, err := client.GetUsersOnlineStatus(context.Background(), req)
+		if err != nil {
+			log.NewError(params.OperationID, "GetUsersOnlineStatus rpc  err", req.String(), err.Error())
+			continue
+		} else {
+			if reply.ErrCode == 0 {
+				wsResult = append(wsResult, reply.SuccessResult...)
+			}
+		}
+	}
+	log.NewInfo(params.OperationID, "call GetUsersOnlineStatus rpc server is success", wsResult)
+	//Online data merge of each node
+	for _, v1 := range params.UserIDList {
+		flag = false
+		temp := new(pbRelay.GetUsersOnlineStatusResp_SuccessResult)
+		for _, v2 := range wsResult {
+			if v2.UserID == v1 {
+				flag = true
+				temp.UserID = v1
+				temp.Status = constant.OnlineStatus
+				temp.DetailPlatformStatus = append(temp.DetailPlatformStatus, v2.DetailPlatformStatus...)
+			}
+
+		}
+		if !flag {
+			temp.UserID = v1
+			temp.Status = constant.OfflineStatus
+		}
+		respResult = append(respResult, temp)
+	}
+	resp := api.GetUsersOnlineStatusResp{CommResp: api.CommResp{ErrCode: 0, ErrMsg: ""}}
+	resp.SuccessResult = respResult
+	log.NewInfo(req.OperationID, "GetUsersOnlineStatus api return", resp)
+	c.JSON(http.StatusOK, resp)
+
+}

internal/api/user/user.go

@@ -14,68 +14,6 @@ import (
 	"strings"
 )
 
-//
-//func GetUsersOnlineStatus(c *gin.Context) {
-//	params := api.GetUsersOnlineStatusReq{}
-//	if err := c.BindJSON(&params); err != nil {
-//		log.NewError(params.OperationID, "bind json failed ", err.Error(), c)
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
-//		return
-//	}
-//
-//	if params.Secret != config.Config.Secret {
-//		log.NewError(params.OperationID, "parse token failed ", params.Secret, config.Config.Secret)
-//		c.JSON(http.StatusBadRequest, gin.H{"errCode": 401, "errMsg": "secret failed"})
-//		return
-//	}
-//
-//	req := &pbRelay.GetUsersOnlineStatusReq{
-//		OperationID: params.OperationID,
-//		UserIDList:  params.UserIDList,
-//	}
-//	var wsResult []*rpc.GetUsersOnlineStatusResp_SuccessResult
-//	var respResult []*rpc.GetUsersOnlineStatusResp_SuccessResult
-//	flag := false
-//	log.NewDebug(params.OperationID, "GetUsersOnlineStatus req come here", params.UserIDList)
-//
-//	grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
-//	for _, v := range grpcCons {
-//		client := rpc.NewOnlineMessageRelayServiceClient(v)
-//		reply, err := client.GetUsersOnlineStatus(context.Background(), req)
-//		if err != nil {
-//			log.NewError(params.OperationID, "GetUsersOnlineStatus rpc  err", req.String(), err.Error())
-//			continue
-//		} else {
-//			if reply.ErrCode == 0 {
-//				wsResult = append(wsResult, reply.SuccessResult...)
-//			}
-//		}
-//	}
-//	log.NewDebug(params.OperationID, "call GetUsersOnlineStatus rpc server is success", wsResult)
-//	//Online data merge of each node
-//	for _, v1 := range params.UserIDList {
-//		flag = false
-//		temp := new(pbRelay.GetUsersOnlineStatusResp_SuccessResult)
-//		for _, v2 := range wsResult {
-//			if v2.UserID == v1 {
-//				flag = true
-//				temp.UserID = v1
-//				temp.Status = constant.OnlineStatus
-//				temp.DetailPlatformStatus = append(temp.DetailPlatformStatus, v2.DetailPlatformStatus...)
-//			}
-//		}
-//		if !flag {
-//			temp.UserID = v1
-//			temp.Status = constant.OfflineStatus
-//		}
-//		respResult = append(respResult, temp)
-//	}
-//	log.NewDebug(params.OperationID, "Finished merged data", respResult)
-//	resp := gin.H{"errCode": 0, "errMsg": "", "data": respResult}
-//
-//	c.JSON(http.StatusOK, resp)
-//}
-
 func GetUserInfo(c *gin.Context) {
 	params := api.GetUserInfoReq{}
 	if err := c.BindJSON(&params); err != nil {
@@ -83,7 +21,7 @@ func GetUserInfo(c *gin.Context) {
 		return
 	}
 	req := &rpc.GetUserInfoReq{}
-	utils.CopyStructFields(&req, params)
+	utils.CopyStructFields(req, &params)
 	var ok bool
 	ok, req.OpUserID = token_verify.GetUserIDFromToken(c.Request.Header.Get("token"))
 	if !ok {

internal/msg_gateway/gate/logic.go

@@ -151,7 +151,6 @@ func (ws *WServer) pullMsgResp(conn *UserConn, m *Req, pb *sdk_ws.PullMessageRes
 	}
 	c, err := proto.Marshal(&mReplyData)
 	log.NewInfo(m.OperationID, "test info is ", len(mReplyData.SingleUserMsg), mReplyData.SingleUserMsg)
-
 	mReply := Resp{
 		ReqIdentifier: m.ReqIdentifier,
 		MsgIncr:       m.MsgIncr,
@@ -162,7 +161,6 @@ func (ws *WServer) pullMsgResp(conn *UserConn, m *Req, pb *sdk_ws.PullMessageRes
 	}
 	log.NewInfo(m.OperationID, "pullMsgResp all data  is ", mReply.ReqIdentifier, mReply.MsgIncr, mReply.ErrCode, mReply.ErrMsg,
 		len(mReply.Data))
-
 	ws.sendMsg(conn, mReply)
 
 }

internal/msg_gateway/gate/rpc_server.go

@@ -4,6 +4,7 @@ import (
 	"Open_IM/pkg/common/config"
 	"Open_IM/pkg/common/constant"
 	"Open_IM/pkg/common/log"
+	"Open_IM/pkg/common/token_verify"
 	"Open_IM/pkg/grpc-etcdv3/getcdv3"
 	pbRelay "Open_IM/pkg/proto/relay"
 	"Open_IM/pkg/utils"
@@ -105,7 +106,11 @@ func (r *RPCServer) OnlinePushMsg(_ context.Context, in *pbRelay.OnlinePushMsgRe
 	}, nil
 }
 func (r *RPCServer) GetUsersOnlineStatus(_ context.Context, req *pbRelay.GetUsersOnlineStatusReq) (*pbRelay.GetUsersOnlineStatusResp, error) {
-	log.NewDebug(req.OperationID, "rpc GetUsersOnlineStatus arrived server", req.String())
+	log.NewInfo(req.OperationID, "rpc GetUsersOnlineStatus arrived server", req.String())
+	if !token_verify.IsMangerUserID(req.OpUserID) {
+		log.NewError(req.OperationID, "no permission GetUsersOnlineStatus ", req.OpUserID)
+		return &pbRelay.GetUsersOnlineStatusResp{ErrCode: constant.ErrAccess.ErrCode, ErrMsg: constant.ErrAccess.ErrMsg}, nil
+	}
 	var resp pbRelay.GetUsersOnlineStatusResp
 	for _, userID := range req.UserIDList {
 		platformList := genPlatformArray()
@@ -125,6 +130,7 @@ func (r *RPCServer) GetUsersOnlineStatus(_ context.Context, req *pbRelay.GetUser
 			resp.SuccessResult = append(resp.SuccessResult, temp)
 		}
 	}
+	log.NewInfo(req.OperationID, "GetUsersOnlineStatus rpc return ", resp.String())
 	return &resp, nil
 }
 func sendMsgToUser(conn *UserConn, bMsg []byte, in *pbRelay.OnlinePushMsgReq, RecvPlatForm, RecvID string) (ResultCode int64) {

internal/rpc/msg/send_msg.go

@@ -141,10 +141,12 @@ func (rpc *rpcChat) SendMsg(_ context.Context, pb *pbChat.SendMsgReq) (*pbChat.S
 				return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
 			}
 		}
-		err2 := rpc.sendMsgToKafka(&msgToMQ, msgToMQ.MsgData.SendID)
-		if err2 != nil {
-			log.NewError(msgToMQ.OperationID, "kafka send msg err:SendID", msgToMQ.MsgData.SendID, msgToMQ.String())
-			return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
+		if msgToMQ.MsgData.SendID != msgToMQ.MsgData.RecvID { //Filter messages sent to yourself
+			err2 := rpc.sendMsgToKafka(&msgToMQ, msgToMQ.MsgData.SendID)
+			if err2 != nil {
+				log.NewError(msgToMQ.OperationID, "kafka send msg err:SendID", msgToMQ.MsgData.SendID, msgToMQ.String())
+				return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
+			}
 		}
 		return returnMsg(&replay, pb, 0, "", msgToMQ.MsgData.ServerMsgID, msgToMQ.MsgData.SendTime)
 	case constant.GroupChatType: