update

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

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) {

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 {
+				//if isSenderSync {
-					pbSaveData.ContentType = constant.SyncSenderMsg
+				//	pbSaveData.ContentType = constant.SyncSenderMsg
-					log.WarnByKv("SyncSenderMsg come here", pbData.OperationID, pbSaveData.String())
+				//	log.WarnByKv("SyncSenderMsg come here", pbData.OperationID, pbSaveData.String())
-					sendMessageToPush(&pbSaveData)
+				//	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, " ")
-		etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
+		saveUserChat(uidAndGroupID[0], &pbSaveData)
-		client := pb.NewGroupClient(etcdConn)
-		req := &pb.GetGroupInfoReq{
-			GroupID:     pbSaveData.RecvID,
-			Token:       pbData.Token,
-			OperationID: pbSaveData.OperationID,
-		}
-		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(uidAndGroupID[0], pbData.ForceList) {
-				if utils.IsContain(v.UserID, pbData.ForceList) {
 			pbSaveData.IsEmphasize = true
 		}
 		sendMessageToPush(&pbSaveData)
-			} else {
+		log.InfoByKv("msg_transfer handle topic success...", "", "")
-				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 }

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)
+}

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()
-	}
-}

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)
+	switch m.SessionType {
-	if err != nil {
+	case constant.SingleChatType:
-		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)
+		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,
 		}
-	pid, offset, err = producer.SendMessage(m, m.RecvID)
+		reply, err := client.GetGroupAllMember(context.Background(), req)
 		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)
+			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)
+	}
+
 }