diff --git a/src/msg_transfer/logic/db.go b/src/msg_transfer/logic/db.go index 93dccfc2b..99aa11474 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.SendTime, pbMsg) + return db.DB.SaveUserChat(uid, 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 8ef4098ce..827b89481 100644 --- a/src/msg_transfer/logic/history_msg_handler.go +++ b/src/msg_transfer/logic/history_msg_handler.go @@ -6,6 +6,7 @@ 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" @@ -54,8 +55,10 @@ 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) + log.Info("", "", "msg_transfer chat type = SingleChatType", isHistory, isPersist, isSenderSync) if isHistory { if msgKey == pbSaveData.RecvID { err := saveUserChat(pbData.RecvID, &pbSaveData) @@ -70,30 +73,58 @@ 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") - 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 + + 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, } - sendMessageToPush(&pbSaveData) - log.InfoByKv("msg_transfer handle topic success...", "", "") + 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) + } + + } + + } + } 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 84521c7bb..b97e3217f 100644 --- a/src/push/content_struct/content.go +++ b/src/push/content_struct/content.go @@ -6,9 +6,7 @@ */ package content_struct -import ( - "encoding/json" -) +import "encoding/json" type Content struct { IsDisplay int32 `json:"isDisplay"` @@ -25,36 +23,3 @@ 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 a7be054b8..adcd92964 100644 --- a/src/push/logic/init.go +++ b/src/push/logic/init.go @@ -8,9 +8,12 @@ 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 ( @@ -32,5 +35,32 @@ 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 3bf685f67..d53f459e2 100644 --- a/src/push/logic/push_to_client.go +++ b/src/push/logic/push_to_client.go @@ -11,7 +11,6 @@ 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" @@ -34,25 +33,19 @@ type EChatContent struct { func MsgToUser(sendPbData *pbRelay.MsgToUserReq, OfflineInfo, Options string) { var wsResult []*pbRelay.SingleMsgToUser isShouldOfflinePush := true - MOptions := utils.JsonStringToMap(Options) //Control whether to push message to sender's other terminal - //isSenderSync := utils.GetSwitchFromOptions(MOptions, "senderSync") + MOptions := utils.JsonStringToMap(Options) 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...) } } - log.InfoByKv("push_result", sendPbData.OperationID, "result", wsResult) - if isOfflinePush { + if isOfflinePush && sendPbData.ContentType != constant.SyncSenderMsg { for _, t := range pushTerminal { for _, v := range wsResult { @@ -108,41 +101,12 @@ func MsgToUser(sendPbData *pbRelay.MsgToUserReq, OfflineInfo, Options string) { func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) { m.MsgID = rpcChat.GetMsgID(m.SendID) - 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) + pid, offset, err := producer.SendMessage(m, m.SendID) if err != nil { - log.ErrorByKv("kafka send failed", m.OperationID, "send data", m.String(), "pid", pid, "offset", offset, "err", err.Error(), flag, key) + 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) } - }