Open-IM-Server/internal/push/logic/push_to_client.go

/*
** description("").
** copyright('open-im,www.open-im.io').
** author("fg,Gordon@open-im.io").
** time(2021/3/5 14:31).
 */
package logic

import (
	push "Open_IM/internal/push/jpush"
	rpcChat "Open_IM/internal/rpc/chat"
	"Open_IM/pkg/common/config"
	"Open_IM/pkg/common/constant"
	"Open_IM/pkg/common/log"
	"Open_IM/pkg/grpc-etcdv3/getcdv3"
	pbChat "Open_IM/pkg/proto/chat"
	pbGroup "Open_IM/pkg/proto/group"
	pbRelay "Open_IM/pkg/proto/relay"
	"Open_IM/pkg/utils"
	"context"
	"encoding/json"
	"strings"
)

type OpenIMContent struct {
	SessionType int    `json:"sessionType"`
	From        string `json:"from"`
	To          string `json:"to"`
	Seq         int64  `json:"seq"`
}
type AtContent struct {
	Text       string   `json:"text"`
	AtUserList []string `json:"atUserList"`
	IsAtSelf   bool     `json:"isAtSelf"`
}

func MsgToUser(sendPbData *pbRelay.MsgToUserReq, OfflineInfo string, Options map[string]int32) {
	var wsResult []*pbRelay.SingleMsgToUser
	isOfflinePush := utils.GetSwitchFromOptions(Options, "offlinePush")
	log.InfoByKv("Get chat from msg_transfer And push chat", sendPbData.OperationID, "PushData", sendPbData, Options, isOfflinePush)
	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)
			continue
		}
		if reply != nil && reply.Resp != nil {
			wsResult = append(wsResult, reply.Resp...)
		}
	}
	log.InfoByKv("push_result", sendPbData.OperationID, "result", wsResult, "sendData", sendPbData)
	if sendPbData.ContentType != constant.Typing && sendPbData.ContentType != constant.HasReadReceipt {
		if isOfflinePush {
			for _, v := range wsResult {
				if v.ResultCode == 0 {
					continue
				}
				//supported terminal
				for _, t := range pushTerminal {
					if v.RecvPlatFormID == t {
						//Use offline push messaging
						var UIDList []string
						UIDList = append(UIDList, v.RecvID)
						customContent := OpenIMContent{
							SessionType: int(sendPbData.SessionType),
							From:        sendPbData.SendID,
							To:          sendPbData.RecvID,
							Seq:         sendPbData.RecvSeq,
						}
						bCustomContent, _ := json.Marshal(customContent)
						jsonCustomContent := string(bCustomContent)
						var content string
						switch sendPbData.ContentType {
						case constant.Text:
							content = constant.ContentType2PushContent[constant.Text]
						case constant.Picture:
							content = constant.ContentType2PushContent[constant.Picture]
						case constant.Voice:
							content = constant.ContentType2PushContent[constant.Voice]
						case constant.Video:
							content = constant.ContentType2PushContent[constant.Video]
						case constant.File:
							content = constant.ContentType2PushContent[constant.File]
						case constant.AtText:
							a := AtContent{}
							_ = utils.JsonStringToStruct(sendPbData.Content, &a)
							if utils.IsContain(v.RecvID, a.AtUserList) {
								content = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
							} else {
								content = constant.ContentType2PushContent[constant.GroupMsg]
							}
						default:
							content = constant.ContentType2PushContent[constant.Common]
						}
						pushResult, err := push.JGAccountListPush(UIDList, content, jsonCustomContent, constant.PlatformIDToName(t))
						if err != nil {
							log.NewError(sendPbData.OperationID, "offline push error", sendPbData.String(), err.Error(), constant.PlatformIDToName(t))
						} else {
							log.NewDebug(sendPbData.OperationID, "offline push return result is ", string(pushResult), sendPbData, constant.PlatformIDToName(t))
						}

					}
				}
			}

		}
	}

}

func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) {
	m.MsgID = rpcChat.GetMsgID(m.SendID)
	m.ClientMsgID = m.MsgID
	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 i, v := range reply.MemberList {
			m.RecvID = v.UserId + " " + groupID
			sendMsgToKafka(m, utils.IntToString(i), "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)
	}

}