From 185c220429022a7ea0b82c38ccaa898bf1f6b585 Mon Sep 17 00:00:00 2001 From: away <1819625428@qq.com> Date: Mon, 28 Jun 2021 16:24:43 +0800 Subject: [PATCH] update --- src/common/config/config.go | 2 +- src/msg_transfer/logic/db.go | 2 +- src/msg_transfer/logic/history_msg_handler.go | 61 +++++-------------- src/push/content_struct/content.go | 37 ++++++++++- src/push/logic/init.go | 30 --------- src/push/logic/push_to_client.go | 54 +++++++++++++--- 6 files changed, 98 insertions(+), 88 deletions(-) diff --git a/src/common/config/config.go b/src/common/config/config.go index d91922dae..d869ebad0 100644 --- a/src/common/config/config.go +++ b/src/common/config/config.go @@ -143,7 +143,7 @@ type config struct { } func init() { - bytes, err := ioutil.ReadFile("../config/config.yaml") + bytes, err := ioutil.ReadFile("config/config.yaml") if err != nil { panic(err) return diff --git a/src/msg_transfer/logic/db.go b/src/msg_transfer/logic/db.go index 99aa11474..93dccfc2b 100644 --- a/src/msg_transfer/logic/db.go +++ b/src/msg_transfer/logic/db.go @@ -12,7 +12,7 @@ func saveUserChat(uid string, pbMsg *pbMsg.MsgSvrToPushSvrChatMsg) error { return err } pbMsg.RecvSeq = seq - return db.DB.SaveUserChat(uid, pbMsg) + return db.DB.SaveUserChat(uid, pbMsg.SendTime, pbMsg) } func getGroupList(groupID string) ([]string, error) { diff --git a/src/msg_transfer/logic/history_msg_handler.go b/src/msg_transfer/logic/history_msg_handler.go index 827b89481..8ef4098ce 100644 --- a/src/msg_transfer/logic/history_msg_handler.go +++ b/src/msg_transfer/logic/history_msg_handler.go @@ -6,7 +6,6 @@ import ( kfk "Open_IM/src/common/kafka" "Open_IM/src/common/log" pbMsg "Open_IM/src/proto/chat" - pb "Open_IM/src/proto/group" pbPush "Open_IM/src/proto/push" "Open_IM/src/utils" "context" @@ -55,10 +54,8 @@ func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string) isHistory := utils.GetSwitchFromOptions(Options, "history") //Control whether to store history messages (mysql) isPersist := utils.GetSwitchFromOptions(Options, "persistent") - //Control whether to push message to sender's other terminal - isSenderSync := utils.GetSwitchFromOptions(Options, "senderSync") if pbData.SessionType == constant.SingleChatType { - log.Info("", "", "msg_transfer chat type = SingleChatType", isHistory, isPersist, isSenderSync) + log.Info("", "", "msg_transfer chat type = SingleChatType", isHistory, isPersist) if isHistory { if msgKey == pbSaveData.RecvID { err := saveUserChat(pbData.RecvID, &pbSaveData) @@ -73,58 +70,30 @@ func (mc *HistoryConsumerHandler) handleChatWs2Mongo(msg []byte, msgKey string) if err != nil { log.ErrorByKv("data insert to mongo err", pbSaveData.OperationID, "data", pbSaveData.String(), "err", err.Error()) } - if isSenderSync { - pbSaveData.ContentType = constant.SyncSenderMsg - log.WarnByKv("SyncSenderMsg come here", pbData.OperationID, pbSaveData.String()) - sendMessageToPush(&pbSaveData) - } + //if isSenderSync { + // pbSaveData.ContentType = constant.SyncSenderMsg + // log.WarnByKv("SyncSenderMsg come here", pbData.OperationID, pbSaveData.String()) + // sendMessageToPush(&pbSaveData) + //} } } + log.InfoByKv("msg_transfer handle topic success...", "", "") } else if pbData.SessionType == constant.GroupChatType { log.Info("", "", "msg_transfer chat type = GroupChatType") - - etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName) - client := pb.NewGroupClient(etcdConn) - req := &pb.GetGroupInfoReq{ - GroupID: pbSaveData.RecvID, - Token: pbData.Token, - OperationID: pbSaveData.OperationID, + uidAndGroupID := strings.Split(pbData.RecvID, " ") + saveUserChat(uidAndGroupID[0], &pbSaveData) + pbSaveData.Options = pbData.Options + pbSaveData.OfflineInfo = pbData.OfflineInfo + if utils.IsContain(uidAndGroupID[0], pbData.ForceList) { + pbSaveData.IsEmphasize = true } - log.Info("", "", "msg_transfer call group rpc, data = %s", req.String()) - reply, err := client.GetGroupInfo(context.Background(), req) - if err != nil { - log.Error("", "", "msg_transfer client.GetGroupInfo fail, err = %s", err.Error()) - return - } - for _, v := range reply.GroupMemberList { - //Store RecvID is userID+" "+groupID when chatType is Group - pbSaveData.RecvID = v.UserID + " " + pbSaveData.RecvID - if isHistory { - saveUserChat(v.UserID, &pbSaveData) - } - pbSaveData.Options = pbData.Options - pbSaveData.OfflineInfo = pbData.OfflineInfo - if v.UserID != pbSaveData.SendID { - if utils.IsContain(v.UserID, pbData.ForceList) { - pbSaveData.IsEmphasize = true - } - sendMessageToPush(&pbSaveData) - } else { - if isSenderSync { - pbSaveData.ContentType = constant.SyncSenderMsg - sendMessageToPush(&pbSaveData) - } - - } - - } - + sendMessageToPush(&pbSaveData) + log.InfoByKv("msg_transfer handle topic success...", "", "") } else { log.Error("", "", "msg_transfer recv chat err, chat.MsgFrom = %d", pbData.SessionType) } - log.InfoByKv("msg_transfer handle topic success...", "", "") } func (HistoryConsumerHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil } diff --git a/src/push/content_struct/content.go b/src/push/content_struct/content.go index b97e3217f..84521c7bb 100644 --- a/src/push/content_struct/content.go +++ b/src/push/content_struct/content.go @@ -6,7 +6,9 @@ */ package content_struct -import "encoding/json" +import ( + "encoding/json" +) type Content struct { IsDisplay int32 `json:"isDisplay"` @@ -23,3 +25,36 @@ func (c *Content) contentToString() string { dataString := string(data) return dataString } + +type groupMemberFullInfo struct { + GroupId string `json:"groupID"` + UserId string `json:"userId"` + Role int `json:"role"` + JoinTime uint64 `json:"joinTime"` + NickName string `json:"nickName"` + FaceUrl string `json:"faceUrl"` +} + +type AgreeOrRejectGroupMember struct { + GroupId string `json:"groupID"` + UserId string `json:"userId"` + Role int `json:"role"` + JoinTime uint64 `json:"joinTime"` + NickName string `json:"nickName"` + FaceUrl string `json:"faceUrl"` + Reason string `json:"reason"` +} + +type CreateGroupSysMsg struct { + uIdCreator string `creatorUid` + initMemberList []groupMemberFullInfo `json: initMemberList` + CreateTime uint64 `json:"CreateTime"` + Text string `json:"text"` +} + +func NewCreateGroupSysMsgString(create *CreateGroupSysMsg, text string) string { + create.Text = text + jstring, _ := json.Marshal(create) + + return string(jstring) +} diff --git a/src/push/logic/init.go b/src/push/logic/init.go index adcd92964..a7be054b8 100644 --- a/src/push/logic/init.go +++ b/src/push/logic/init.go @@ -8,12 +8,9 @@ package logic import ( "Open_IM/src/common/config" - "Open_IM/src/common/db" - "Open_IM/src/common/db/mysql_model/im_mysql_model" "Open_IM/src/common/kafka" "Open_IM/src/common/log" "Open_IM/src/utils" - "time" ) var ( @@ -35,32 +32,5 @@ func init() { func Run() { go rpcServer.run() - go scheduleDelete() go pushCh.pushConsumerGroup.RegisterHandleAndConsumer(&pushCh) } - -func scheduleDelete() { - //uid, _ := im_mysql_model.SelectAllUID() - //db.DB.DelHistoryChat(0, uid) - //log.Info("", "", "sssssssssss") - //if err != nil { - // db.DB.DelHistoryChat(0, uid) - //} - - for { - now := time.Now() - // 计算下一个零点 - next := now.Add(time.Hour * 24) - next = time.Date(next.Year(), next.Month(), next.Day(), 0, 0, 0, 0, next.Location()) - t := time.NewTimer(next.Sub(now)) - <-t.C - - uid, err := im_mysql_model.SelectAllUID() - if err != nil { - db.DB.DelHistoryChat(int64(config.Config.Mongo.DBRetainChatRecords), uid) - } - - //以下为定时执行的操作 - scheduleDelete() - } -} diff --git a/src/push/logic/push_to_client.go b/src/push/logic/push_to_client.go index d53f459e2..3bf685f67 100644 --- a/src/push/logic/push_to_client.go +++ b/src/push/logic/push_to_client.go @@ -11,6 +11,7 @@ import ( "Open_IM/src/common/constant" "Open_IM/src/common/log" pbChat "Open_IM/src/proto/chat" + pbGroup "Open_IM/src/proto/group" pbRelay "Open_IM/src/proto/relay" pbGetInfo "Open_IM/src/proto/user" rpcChat "Open_IM/src/rpc/chat/chat" @@ -33,19 +34,25 @@ type EChatContent struct { func MsgToUser(sendPbData *pbRelay.MsgToUserReq, OfflineInfo, Options string) { var wsResult []*pbRelay.SingleMsgToUser isShouldOfflinePush := true - MOptions := utils.JsonStringToMap(Options) + MOptions := utils.JsonStringToMap(Options) //Control whether to push message to sender's other terminal + //isSenderSync := utils.GetSwitchFromOptions(MOptions, "senderSync") isOfflinePush := utils.GetSwitchFromOptions(MOptions, "offlinePush") log.InfoByKv("Get chat from msg_transfer And push chat", sendPbData.OperationID, "PushData", sendPbData) grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName) //Online push message + log.InfoByKv("test", sendPbData.OperationID, "len grpc", len(grpcCons), "data", sendPbData) for _, v := range grpcCons { msgClient := pbRelay.NewOnlineMessageRelayServiceClient(v) reply, err := msgClient.MsgToUser(context.Background(), sendPbData) + if err != nil { + log.InfoByKv("push data to client rpc err", sendPbData.OperationID, "err", err) + } if reply != nil && reply.Resp != nil && err == nil { wsResult = append(wsResult, reply.Resp...) } } - if isOfflinePush && sendPbData.ContentType != constant.SyncSenderMsg { + log.InfoByKv("push_result", sendPbData.OperationID, "result", wsResult) + if isOfflinePush { for _, t := range pushTerminal { for _, v := range wsResult { @@ -101,12 +108,41 @@ func MsgToUser(sendPbData *pbRelay.MsgToUserReq, OfflineInfo, Options string) { func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) { m.MsgID = rpcChat.GetMsgID(m.SendID) - pid, offset, err := producer.SendMessage(m, m.SendID) - if err != nil { - log.ErrorByKv("sys send msg to kafka failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), "msgKey--sendID", m.SendID) - } - pid, offset, err = producer.SendMessage(m, m.RecvID) - if err != nil { - log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), "msgKey--recvID", m.RecvID) + switch m.SessionType { + case constant.SingleChatType: + sendMsgToKafka(m, m.SendID, "msgKey--sendID") + sendMsgToKafka(m, m.RecvID, "msgKey--recvID") + 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: m.RecvID, + Token: config.Config.Secret, + OperationID: m.OperationID, + } + reply, err := client.GetGroupAllMember(context.Background(), req) + if err != nil { + log.Error(m.Token, m.OperationID, "rpc getGroupInfo failed, err = %s", err.Error()) + return + } + if reply.ErrorCode != 0 { + log.Error(m.Token, m.OperationID, "rpc getGroupInfo failed, err = %s", reply.ErrorMsg) + return + } + groupID := m.RecvID + for _, v := range reply.MemberList { + m.RecvID = v.UserId + " " + groupID + sendMsgToKafka(m, m.RecvID, "msgKey--recvID+\" \"+groupID") + } + default: + } + +} +func sendMsgToKafka(m *pbChat.WSToMsgSvrChatMsg, key string, flag string) { + pid, offset, err := producer.SendMessage(m, key) + if err != nil { + log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), flag, key) + } + }