IM/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2026-05-19 00:09:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

add cmd/rpc

Browse Source
This commit is contained in:
xmcy0011
2021-10-11 21:51:37 +08:00
parent 1a12eb9703
commit 585715adbf
42 changed files with 20 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/rpc/chat/pull_message.go
+131
View File
@@ -0,0 +1,131 @@
package chat
import (
"context"
"github.com/garyburd/redigo/redis"
commonDB "Open_IM/src/common/db"
"Open_IM/src/common/log"
"sort"
"strings"
pbMsg "Open_IM/src/proto/chat"
)
func (rpc *rpcChat) GetNewSeq(_ context.Context, in *pbMsg.GetNewSeqReq) (*pbMsg.GetNewSeqResp, error) {
log.InfoByKv("rpc getNewSeq is arriving", in.OperationID, in.String())
//seq, err := model.GetBiggestSeqFromReceive(in.UserID)
seq, err := commonDB.DB.GetUserSeq(in.UserID)
resp := new(pbMsg.GetNewSeqResp)
if err == nil {
resp.Seq = seq
resp.ErrCode = 0
resp.ErrMsg = ""
return resp, err
} else {
if err == redis.ErrNil {
resp.Seq = 0
} else {
log.ErrorByKv("getSeq from redis error", in.OperationID, "args", in.String(), "err", err.Error())
resp.Seq = -1
}
resp.ErrCode = 0
resp.ErrMsg = ""
return resp, nil
}
}
func (rpc *rpcChat) PullMessage(_ context.Context, in *pbMsg.PullMessageReq) (*pbMsg.PullMessageResp, error) {
log.InfoByKv("rpc pullMessage is arriving", in.OperationID, "args", in.String())
resp := new(pbMsg.PullMessageResp)
var respSingleMsgFormat []*pbMsg.GatherFormat
var respGroupMsgFormat []*pbMsg.GatherFormat
SingleMsgFormat, GroupMsgFormat, MaxSeq, MinSeq, err := commonDB.DB.GetUserChat(in.UserID, in.SeqBegin, in.SeqEnd)
if err != nil {
log.ErrorByKv("pullMsg data error", in.OperationID, in.String())
resp.ErrCode = 1
resp.ErrMsg = err.Error()
return resp, nil
}
respSingleMsgFormat = singleMsgHandleByUser(SingleMsgFormat, in.UserID)
respGroupMsgFormat = groupMsgHandleByUser(GroupMsgFormat)
return &pbMsg.PullMessageResp{
ErrCode: 0,
ErrMsg: "",
MaxSeq: MaxSeq,
MinSeq: MinSeq,
SingleUserMsg: respSingleMsgFormat,
GroupUserMsg: respGroupMsgFormat,
}, nil
}
func singleMsgHandleByUser(allMsg []*pbMsg.MsgFormat, ownerId string) []*pbMsg.GatherFormat {
var userid string
var respMsgFormat []*pbMsg.GatherFormat
m := make(map[string]MsgFormats)
//Gather messages in the dimension of users
for _, v := range allMsg {
if v.RecvID != ownerId {
userid = v.RecvID
} else {
userid = v.SendID
}
if value, ok := m[userid]; !ok {
var t MsgFormats
m[userid] = append(t, v)
} else {
m[userid] = append(value, v)
}
}
//Return in pb format
for user, msg := range m {
tempUserMsg := new(pbMsg.GatherFormat)
tempUserMsg.ID = user
tempUserMsg.List = msg
sort.Sort(msg)
respMsgFormat = append(respMsgFormat, tempUserMsg)
}
return respMsgFormat
}
func groupMsgHandleByUser(allMsg []*pbMsg.MsgFormat) []*pbMsg.GatherFormat {
var respMsgFormat []*pbMsg.GatherFormat
m := make(map[string]MsgFormats)
//Gather messages in the dimension of users
for _, v := range allMsg {
//Get group ID
groupID := strings.Split(v.RecvID, " ")[1]
if value, ok := m[groupID]; !ok {
var t MsgFormats
m[groupID] = append(t, v)
} else {
m[groupID] = append(value, v)
}
}
//Return in pb format
for groupID, msg := range m {
tempUserMsg := new(pbMsg.GatherFormat)
tempUserMsg.ID = groupID
tempUserMsg.List = msg
sort.Sort(msg)
respMsgFormat = append(respMsgFormat, tempUserMsg)
}
return respMsgFormat
}
type MsgFormats []*pbMsg.MsgFormat
// Implement the sort.Interface interface to get the number of elements method
func (s MsgFormats) Len() int {
return len(s)
}
//Implement the sort.Interface interface comparison element method
func (s MsgFormats) Less(i, j int) bool {
return s[i].SendTime < s[j].SendTime
}
//Implement the sort.Interface interface exchange element method
func (s MsgFormats) Swap(i, j int) {
s[i], s[j] = s[j], s[i]
}
internal/rpc/chat/rpcChat.go
+65
View File
@@ -0,0 +1,65 @@
package chat
import (
"Open_IM/src/common/config"
"Open_IM/src/common/kafka"
log2 "Open_IM/src/common/log"
"Open_IM/src/grpc-etcdv3/getcdv3"
pbChat "Open_IM/src/proto/chat"
"Open_IM/src/utils"
"google.golang.org/grpc"
"net"
"strconv"
"strings"
)
type rpcChat struct {
rpcPort int
rpcRegisterName string
etcdSchema string
etcdAddr []string
producer *kafka.Producer
}
func NewRpcChatServer(port int) *rpcChat {
rc := rpcChat{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImOfflineMessageName,
etcdSchema: config.Config.Etcd.EtcdSchema,
etcdAddr: config.Config.Etcd.EtcdAddr,
}
rc.producer = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.Ws2mschat.Topic)
return &rc
}
func (rpc *rpcChat) Run() {
log2.Info("", "", "rpc get_token init...")
address := utils.ServerIP + ":" + strconv.Itoa(rpc.rpcPort)
listener, err := net.Listen("tcp", address)
if err != nil {
log2.Error("", "", "listen network failed, err = %s, address = %s", err.Error(), address)
return
}
log2.Info("", "", "listen network success, address = %s", address)
//grpc server
srv := grpc.NewServer()
defer srv.GracefulStop()
//service registers with etcd
pbChat.RegisterChatServer(srv, rpc)
err = getcdv3.RegisterEtcd(rpc.etcdSchema, strings.Join(rpc.etcdAddr, ","), utils.ServerIP, rpc.rpcPort, rpc.rpcRegisterName, 10)
if err != nil {
log2.Error("", "", "register rpc get_token to etcd failed, err = %s", err.Error())
return
}
err = srv.Serve(listener)
if err != nil {
log2.Info("", "", "rpc get_token fail, err = %s", err.Error())
return
}
log2.Info("", "", "rpc get_token init success")
}
internal/rpc/chat/send_msg.go
+191
View File
@@ -0,0 +1,191 @@
package chat
import (
"Open_IM/internal/api/group"
"Open_IM/internal/push/content_struct"
"Open_IM/src/common/config"
"Open_IM/src/common/constant"
http2 "Open_IM/src/common/http"
"Open_IM/src/common/log"
"Open_IM/src/grpc-etcdv3/getcdv3"
pbChat "Open_IM/src/proto/chat"
pbGroup "Open_IM/src/proto/group"
"Open_IM/src/utils"
"context"
"encoding/json"
"math/rand"
"net/http"
"strconv"
"strings"
"time"
)
type MsgCallBackReq struct {
SendID string `json:"sendID"`
RecvID string `json:"recvID"`
Content string `json:"content"`
SendTime int64 `json:"sendTime"`
MsgFrom int32 `json:"msgFrom"`
ContentType int32 `json:"contentType"`
SessionType int32 `json:"sessionType"`
PlatformID int32 `json:"senderPlatformID"`
}
type MsgCallBackResp struct {
ErrCode int32 `json:"errCode"`
ErrMsg string `json:"errMsg"`
ResponseErrCode int32 `json:"responseErrCode"`
ResponseResult struct {
ModifiedMsg string `json:"modifiedMsg"`
Ext string `json:"ext"`
}
}
func (rpc *rpcChat) UserSendMsg(_ context.Context, pb *pbChat.UserSendMsgReq) (*pbChat.UserSendMsgResp, error) {
replay := pbChat.UserSendMsgResp{}
log.InfoByKv("sendMsg", pb.OperationID, "args", pb.String())
if !utils.VerifyToken(pb.Token, pb.SendID) {
return returnMsg(&replay, pb, http.StatusUnauthorized, "token validate err,not authorized", "", 0)
}
serverMsgID := GetMsgID(pb.SendID)
pbData := pbChat.WSToMsgSvrChatMsg{}
pbData.MsgFrom = pb.MsgFrom
pbData.SessionType = pb.SessionType
pbData.ContentType = pb.ContentType
pbData.Content = pb.Content
pbData.RecvID = pb.RecvID
pbData.ForceList = pb.ForceList
pbData.OfflineInfo = pb.OffLineInfo
pbData.Options = pb.Options
pbData.PlatformID = pb.PlatformID
pbData.ClientMsgID = pb.ClientMsgID
pbData.SendID = pb.SendID
pbData.SenderNickName = pb.SenderNickName
pbData.SenderFaceURL = pb.SenderFaceURL
pbData.MsgID = serverMsgID
pbData.OperationID = pb.OperationID
pbData.Token = pb.Token
pbData.SendTime = utils.GetCurrentTimestampByNano()
m := MsgCallBackResp{}
if config.Config.MessageCallBack.CallbackSwitch {
bMsg, err := http2.Post(config.Config.MessageCallBack.CallbackUrl, MsgCallBackReq{
SendID: pb.SendID,
RecvID: pb.RecvID,
Content: pb.Content,
SendTime: pbData.SendTime,
MsgFrom: pbData.MsgFrom,
ContentType: pb.ContentType,
SessionType: pb.SessionType,
PlatformID: pb.PlatformID,
}, "application/json; charset=utf-8")
if err != nil {
log.ErrorByKv("callback to Business server err", pb.OperationID, "args", pb.String(), "err", err.Error())
return returnMsg(&replay, pb, http.StatusInternalServerError, http.StatusText(http.StatusInternalServerError), "", 0)
} else if err = json.Unmarshal(bMsg, &m); err != nil {
log.ErrorByKv("ws json Unmarshal err", pb.OperationID, "args", pb.String(), "err", err.Error())
return returnMsg(&replay, pb, 200, err.Error(), "", 0)
} else {
if m.ErrCode != 0 {
return returnMsg(&replay, pb, m.ResponseErrCode, m.ErrMsg, "", 0)
} else {
pbData.Content = m.ResponseResult.ModifiedMsg
err1 := rpc.sendMsgToKafka(&pbData, pbData.RecvID)
err2 := rpc.sendMsgToKafka(&pbData, pbData.SendID)
if err1 != nil || err2 != nil {
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
}
}
} else {
switch pbData.SessionType {
case constant.SingleChatType:
err1 := rpc.sendMsgToKafka(&pbData, pbData.RecvID)
err2 := rpc.sendMsgToKafka(&pbData, pbData.SendID)
if err1 != nil || err2 != nil {
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
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: pbData.RecvID,
Token: pbData.Token,
OperationID: pbData.OperationID,
}
reply, err := client.GetGroupAllMember(context.Background(), req)
if err != nil {
log.Error(pbData.Token, pbData.OperationID, "rpc send_msg getGroupInfo failed, err = %s", err.Error())
return returnMsg(&replay, pb, 201, err.Error(), "", 0)
}
if reply.ErrorCode != 0 {
log.Error(pbData.Token, pbData.OperationID, "rpc send_msg getGroupInfo failed, err = %s", reply.ErrorMsg)
return returnMsg(&replay, pb, reply.ErrorCode, reply.ErrorMsg, "", 0)
}
var addUidList []string
switch pbData.ContentType {
case constant.KickGroupMemberTip:
var notification content_struct.NotificationContent
var kickContent group.KickGroupMemberReq
err := utils.JsonStringToStruct(pbData.Content, &notification)
if err != nil {
log.ErrorByKv("json unmarshall err", pbData.OperationID, "err", err.Error())
return returnMsg(&replay, pb, 200, err.Error(), "", 0)
} else {
err := utils.JsonStringToStruct(notification.Detail, &kickContent)
if err != nil {
log.ErrorByKv("json unmarshall err", pbData.OperationID, "err", err.Error())
return returnMsg(&replay, pb, 200, err.Error(), "", 0)
}
for _, v := range kickContent.UidListInfo {
addUidList = append(addUidList, v.UserId)
}
}
case constant.QuitGroupTip:
addUidList = append(addUidList, pbData.SendID)
default:
}
groupID := pbData.RecvID
for i, v := range reply.MemberList {
pbData.RecvID = v.UserId + " " + groupID
err := rpc.sendMsgToKafka(&pbData, utils.IntToString(i))
if err != nil {
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
}
for i, v := range addUidList {
pbData.RecvID = v + " " + groupID
err := rpc.sendMsgToKafka(&pbData, utils.IntToString(i+1))
if err != nil {
return returnMsg(&replay, pb, 201, "kafka send msg err", "", 0)
}
}
return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
default:
}
}
return returnMsg(&replay, pb, 203, "unkonwn sessionType", "", 0)
}
func (rpc *rpcChat) sendMsgToKafka(m *pbChat.WSToMsgSvrChatMsg, key string) error {
pid, offset, err := rpc.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(), "key", key)
}
return err
}
func GetMsgID(sendID string) string {
t := time.Now().Format("2006-01-02 15:04:05")
return t + "-" + sendID + "-" + strconv.Itoa(rand.Int())
}
func returnMsg(replay *pbChat.UserSendMsgResp, pb *pbChat.UserSendMsgReq, errCode int32, errMsg, serverMsgID string, sendTime int64) (*pbChat.UserSendMsgResp, error) {
replay.ErrCode = errCode
replay.ErrMsg = errMsg
replay.ReqIdentifier = pb.ReqIdentifier
replay.ClientMsgID = pb.ClientMsgID
replay.ServerMsgID = serverMsgID
replay.SendTime = sendTime
return replay, nil
}