msb-public
/
Open-IM-Server
mirror of https://github.com/OpenIMSDK/Open-IM-Server
2
0
Code Issues Projects Releases Wiki Activity

Merge pull request #173 from OpenIMSDK/tuoyun

Browse Source 
Tuoyun
pull/174/head
Judgelight 4 years ago committed by GitHub
parent 13efbf5f98 9ab9ae0408
commit d3a38963ae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 2007 additions and 3132 deletions
Show Stats Download Patch File Download Diff File

2
internal/api/user/user.go
Unescape View File

@ -45,7 +45,7 @@ func GetUsersInfo(c *gin.Context) {
var publicUserInfoList []*open_im_sdk.PublicUserInfo
for _, v := range RpcResp.UserInfoList {
publicUserInfoList = append(publicUserInfoList,
&open_im_sdk.PublicUserInfo{UserID: v.UserID, Nickname: v.Nickname, FaceURL: v.FaceURL, Gender: v.Gender})
&open_im_sdk.PublicUserInfo{UserID: v.UserID, Nickname: v.Nickname, FaceURL: v.FaceURL, Gender: v.Gender, Ex: v.Ex})
}
resp := api.GetUsersInfoResp{CommResp: api.CommResp{ErrCode: RpcResp.CommonResp.ErrCode, ErrMsg: RpcResp.CommonResp.ErrMsg}, UserInfoList: publicUserInfoList}

3
internal/cms_api/group/group.go
Unescape View File

@ -286,13 +286,12 @@ func GetGroupMembers(c *gin.Context) {
MemberPosition: int(groupMembers.RoleLevel),
MemberNickName: groupMembers.Nickname,
MemberId: groupMembers.UserID,
JoinTime: utils.UnixSecondToTime(groupMembers.JoinTime).String(),
JoinTime: utils.UnixSecondToTime(int64(groupMembers.JoinTime)).String(),
})
}
openIMHttp.RespHttp200(c, constant.OK, resp)
}
func AddGroupMembers(c *gin.Context) {
var (
req cms_api_struct.RemoveGroupMembersRequest

4
internal/msg_gateway/gate/init.go
Unescape View File

@ -23,8 +23,8 @@ func Init(rpcPort, wsPort int) {
log.NewPrivateLog(config.Config.ModuleName.LongConnSvrName)
rwLock = new(sync.RWMutex)
validate = validator.New()
statistics.NewStatistics(&sendMsgCount, config.Config.ModuleName.LongConnSvrName, fmt.Sprintf("%d second recv to msg_gateway sendMsgCount", sendMsgCount), 10)
statistics.NewStatistics(&userCount, config.Config.ModuleName.LongConnSvrName, fmt.Sprintf("%d second add user conn", userCount), 10)
statistics.NewStatistics(&sendMsgCount, config.Config.ModuleName.LongConnSvrName, fmt.Sprintf("%d second recv to msg_gateway sendMsgCount", sendMsgCount), 300)
statistics.NewStatistics(&userCount, config.Config.ModuleName.LongConnSvrName, fmt.Sprintf("%d second add user conn", userCount), 300)
ws.onInit(wsPort)
rpcSvr.onInit(rpcPort)
}

53
internal/msg_gateway/gate/logic.go
Unescape View File

@ -52,6 +52,8 @@ func (ws *WServer) msgParse(conn *UserConn, binaryMsg []byte) {
ws.getSeqReq(conn, &m)
case constant.WSSendMsg:
ws.sendMsgReq(conn, &m)
case constant.WSSendSignalMsg:
ws.sendSignalMsgReq(conn, &m)
case constant.WSPullMsgBySeqList:
ws.pullMsgBySeqListReq(conn, &m)
default:
@ -193,6 +195,57 @@ func (ws *WServer) sendMsgResp(conn *UserConn, m *Req, pb *pbChat.SendMsgResp) {
ws.sendMsg(conn, mReply)
}
func (ws *WServer) sendSignalMsgReq(conn *UserConn, m *Req) {
sendMsgCount++
log.NewInfo(m.OperationID, "Ws call success to sendSignalMsgReq start", m.MsgIncr, m.ReqIdentifier, m.SendID)
nReply := new(pbChat.SendMsgResp)
isPass, errCode, errMsg, pData := ws.argsValidate(m, constant.WSSendMsg)
if isPass {
data := pData.(sdk_ws.MsgData)
pbData := pbChat.SendMsgReq{
Token: m.Token,
OperationID: m.OperationID,
MsgData: &data,
}
log.NewInfo(m.OperationID, "Ws call success to sendSignalMsgReq middle", m.ReqIdentifier, m.SendID, m.MsgIncr, data)
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
client := pbChat.NewChatClient(etcdConn)
reply, err := client.SendMsg(context.Background(), &pbData)
if err != nil {
log.NewError(pbData.OperationID, "rpc sendMsg err", err.Error())
nReply.ErrCode = 200
nReply.ErrMsg = err.Error()
ws.sendSignalMsgResp(conn, m, nReply)
} else {
log.NewInfo(pbData.OperationID, "rpc call success to sendMsgReq", reply.String())
ws.sendSignalMsgResp(conn, m, reply)
}
} else {
nReply.ErrCode = errCode
nReply.ErrMsg = errMsg
ws.sendSignalMsgResp(conn, m, nReply)
}
}
func (ws *WServer) sendSignalMsgResp(conn *UserConn, m *Req, pb *pbChat.SendMsgResp) {
// := make(map[string]interface{})
var mReplyData sdk_ws.UserSendMsgResp
mReplyData.ClientMsgID = pb.GetClientMsgID()
mReplyData.ServerMsgID = pb.GetServerMsgID()
mReplyData.SendTime = pb.GetSendTime()
b, _ := proto.Marshal(&mReplyData)
mReply := Resp{
ReqIdentifier: m.ReqIdentifier,
MsgIncr: m.MsgIncr,
ErrCode: pb.GetErrCode(),
ErrMsg: pb.GetErrMsg(),
OperationID: m.OperationID,
Data: b,
}
ws.sendMsg(conn, mReply)
}
func (ws *WServer) sendMsg(conn *UserConn, mReply interface{}) {
var b bytes.Buffer
enc := gob.NewEncoder(&b)

2
internal/msg_gateway/gate/validate.go
Unescape View File

@ -58,7 +58,7 @@ type SeqListData struct {
func (ws *WServer) argsValidate(m *Req, r int32) (isPass bool, errCode int32, errMsg string, returnData interface{}) {
switch r {
case constant.WSSendMsg:
case constant.WSSendMsg | constant.WSSendSignalMsg:
data := open_im_sdk.MsgData{}
if err := proto.Unmarshal(m.Data, &data); err != nil {
log.ErrorByKv("Decode Data struct err", "", "err", err.Error(), "reqIdentifier", r)

4
internal/msg_transfer/logic/history_msg_handler.go
Unescape View File

@ -26,8 +26,8 @@ type HistoryConsumerHandler struct {
}
func (mc *HistoryConsumerHandler) Init() {
statistics.NewStatistics(&mc.singleMsgCount, config.Config.ModuleName.MsgTransferName, "singleMsgCount insert to mongo ", 10)
statistics.NewStatistics(&mc.groupMsgCount, config.Config.ModuleName.MsgTransferName, "groupMsgCount insert to mongo ", 10)
statistics.NewStatistics(&mc.singleMsgCount, config.Config.ModuleName.MsgTransferName, "singleMsgCount insert to mongo ", 300)
statistics.NewStatistics(&mc.groupMsgCount, config.Config.ModuleName.MsgTransferName, "groupMsgCount insert to mongo ", 300)
mc.msgHandle = make(map[string]fcb)
mc.msgHandle[config.Config.Kafka.Ws2mschat.Topic] = mc.handleChatWs2Mongo

2
internal/push/logic/init.go
Unescape View File

@ -31,7 +31,7 @@ func Init(rpcPort int) {
}
func init() {
producer = kafka.NewKafkaProducer(config.Config.Kafka.Ws2mschat.Addr, config.Config.Kafka.Ws2mschat.Topic)
statistics.NewStatistics(&count, config.Config.ModuleName.PushName, fmt.Sprintf("%d second push to msg_gateway count", 10), 10)
statistics.NewStatistics(&count, config.Config.ModuleName.PushName, fmt.Sprintf("%d second push to msg_gateway count", 300), 300)
}
func Run() {

9
internal/rpc/group/group.go
Unescape View File

@ -79,7 +79,7 @@ func (s *groupServer) CreateGroup(ctx context.Context, req *pbGroup.CreateGroupR
canCreate, err := callbackBeforeCreateGroup(req)
if err != nil || !canCreate {
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "callbackBeforeCreateGroup failed", )
log.NewError(req.OperationID, utils.GetSelfFuncName(), "callbackBeforeCreateGroup failed")
}
}
//Time stamp + MD5 to generate group chat id
@ -274,7 +274,7 @@ func (s *groupServer) GetGroupAllMember(ctx context.Context, req *pbGroup.GetGro
for _, v := range memberList {
var node open_im_sdk.GroupMemberFullInfo
utils.CopyStructFields(&node, &v)
cp.GroupMemberDBCopyOpenIM(&node, &v)
resp.MemberList = append(resp.MemberList, &node)
}
log.NewInfo(req.OperationID, "GetGroupAllMember rpc return ", resp.String())
@ -395,7 +395,7 @@ func (s *groupServer) GetGroupMembersInfo(ctx context.Context, req *pbGroup.GetG
continue
} else {
utils.CopyStructFields(&memberNode, memberInfo)
memberNode.JoinTime = memberInfo.JoinTime.Unix()
memberNode.JoinTime = int32(memberInfo.JoinTime.Unix())
resp.MemberList = append(resp.MemberList, &memberNode)
}
}
@ -827,7 +827,7 @@ func (s *groupServer) GetGroupMembersCMS(_ context.Context, req *pbGroup.GetGrou
GroupID: req.GroupId,
UserID: groupMember.UserID,
RoleLevel: groupMember.RoleLevel,
JoinTime: groupMember.JoinTime.Unix(),
JoinTime: int32(groupMember.JoinTime.Unix()),
Nickname: groupMember.Nickname,
FaceURL: groupMember.FaceURL,
JoinSource: groupMember.JoinSource,
@ -902,7 +902,6 @@ func (s *groupServer) AddGroupMembersCMS(_ context.Context, req *pbGroup.AddGrou
return resp, nil
}
func (s *groupServer) GetUserReqApplicationList(_ context.Context, req *pbGroup.GetUserReqApplicationListReq) (*pbGroup.GetUserReqApplicationListResp, error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp := &pbGroup.GetUserReqApplicationListResp{}

2
pkg/common/constant/constant.go
Unescape View File

@ -20,8 +20,10 @@ const (
WSGetNewestSeq = 1001
WSPullMsgBySeqList = 1002
WSSendMsg = 1003
WSSendSignalMsg = 1004
WSPushMsg = 2001
WSKickOnlineMsg = 2002
WsLogoutMsg = 2003
WSDataError = 3001
///ContentType

2
pkg/common/utils/utils.go
Unescape View File

@ -117,7 +117,7 @@ func GroupMemberDBCopyOpenIM(dst *open_im_sdk.GroupMemberFullInfo, src *db.Group
dst.AppMangerLevel = 1
}
dst.JoinTime = src.JoinTime.Unix()
dst.JoinTime = int32(src.JoinTime.Unix())
return nil
}

17
pkg/grpc-etcdv3/getcdv3/register.go
Unescape View File

@ -38,6 +38,7 @@ func RegisterEtcd4Unique(schema, etcdAddr, myHost string, myPort int, serviceNam
//etcdAddr separated by commas
func RegisterEtcd(schema, etcdAddr, myHost string, myPort int, serviceName string, ttl int) error {
ttl = ttl * 3
cli, err := clientv3.New(clientv3.Config{
Endpoints: strings.Split(etcdAddr, ","), DialTimeout: 5 * time.Second})
@ -72,11 +73,21 @@ func RegisterEtcd(schema, etcdAddr, myHost string, myPort int, serviceName strin
go func() {
for {
select {
case _, ok := <-kresp:
case pv, ok := <-kresp:
if ok == true {
//log.Debug("", "KeepAlive kresp ok", v, schema, etcdAddr, myHost, myPort, serviceName, ttl)
log.Debug("", "KeepAlive kresp ok", pv)
} else {
//log.Error("", "KeepAlive kresp failed", schema, etcdAddr, myHost, myPort, serviceName, ttl)
log.Error("", "KeepAlive kresp failed", pv)
t := time.NewTicker(time.Duration(ttl) * time.Second)
for {
select {
case <-t.C:
}
if _, err := cli.Put(ctx, serviceKey, serviceValue, clientv3.WithLease(resp.ID)); err != nil {
log.Error("", "etcd Put failed ", err.Error(), serviceKey, serviceValue, resp.ID)
}
log.Info("", "etcd Put ok", serviceKey, serviceValue, resp.ID)
}
}
}
}

4548
pkg/proto/sdk_ws/ws.pb.go
View File

File diff suppressed because it is too large Load Diff

118
pkg/proto/sdk_ws/ws.proto
Unescape View File

@ -26,7 +26,7 @@ message GroupMemberFullInfo {
string groupID = 1 ;
string userID = 2 ;
int32 roleLevel = 3;
int64 joinTime = 4;
int32 joinTime = 4;
string nickname = 5;
string faceURL = 6;
int32 appMangerLevel = 7; //if >0
@ -40,6 +40,7 @@ message PublicUserInfo{
string nickname = 2;
string faceURL = 3;
int32 gender = 4;
string ex = 5;
}
message UserInfo{
@ -338,45 +339,110 @@ message ConversationUpdateTips{
}
///callback
message CommonCallbackURLReq {
string CallbackCommand = 1 [json_name = "code"];
string OpenIMServerID = 2;
string OperationID = 3;
///cms
message RequestPagination {
int32 pageNumber = 1;
int32 showNumber = 2;
}
message CommonCallbackURLResp {
string Code = 1 [json_name = "code"];
string Msg = 2 [json_name = "msg"];
string OperationID = 3 [json_name = "operationID"];
message ResponsePagination {
int32 CurrentPage = 5;
int32 ShowNumber = 6;
}
message CallbackBeforeSendMsgReq {
commonReq CommonCallbackURLReq = 1;
///////////////////signal//////////////
message SignalReq {
oneof payload {
SignalInviteReq invite = 1;
SignalInviteInGroupReq inviteInGroup= 2;
SignalCancelReq cancel = 3;
SignalAcceptReq accept = 4;
SignalHungUpReq hungUp = 5;
SignalRejectReq reject = 6;
}
}
message CallbackBeforeSendMsgResp {
commonResp CommonCallbackURLResp = 1;
string FromUserID = 2;
message SignalResp {
oneof payload {
SignalInviteReply invite = 1;
SignalInviteInGroupReply inviteInGroup= 2;
SignalCancelReply cancel = 3;
SignalAcceptReply accept = 4;
SignalHungUpReply hungUp = 5;
SignalRejectReply reject = 6;
}
}
message CallbackAfterAddFriendReq {
commonReq CommonCallbackURLReq = 1;
message InvitationInfo {
string inviterUserID = 1;
repeated string inviteeUserIDList = 2;
string customData = 3;
string groupID = 4;
}
message CallbackAfterAddFriendResp {
commonResp CommonCallbackURLResp = 1;
message SignalInviteReq {
InvitationInfo invitation = 1;
}
message SignalInviteReply {
string token = 1;
string roomID = 2;
string liveURL = 3;
}
///cms
message RequestPagination {
int32 pageNumber = 1;
int32 showNumber = 2;
message SignalInviteInGroupReq {
InvitationInfo invitation = 1;
}
message ResponsePagination {
int32 CurrentPage = 5;
int32 ShowNumber = 6;
message SignalInviteInGroupReply {
string token = 1;
string roomID = 2;
string liveURL = 3;
}
message SignalCancelReq {
string inviterUserID = 1;
SignalInviteReq invitation = 2;
}
message SignalCancelReply {
}
message SignalAcceptReq {
string inviteeUserID = 1;
SignalInviteReq invitation = 2;
}
message SignalAcceptReply {
string token = 1;
string roomID = 2;
string liveURL = 3;
}
message SignalHungUpReq {
string UserID = 1;
SignalInviteReq invitation = 2;
}
message SignalHungUpReply {
}
message SignalRejectReq {
string inviteeUserID = 1;
SignalInviteReq invitation = 2;
}
message SignalRejectReply {
}

7
pkg/statistics/statistics.go
Unescape View File

@ -1,6 +1,7 @@
package statistics
import (
"Open_IM/pkg/common/log"
"time"
)
@ -14,13 +15,13 @@ type Statistics struct {
func (s *Statistics) output() {
t := time.NewTicker(time.Duration(s.SleepTime) * time.Second)
defer t.Stop()
//var sum uint64
var sum uint64
for {
//sum = *s.Count
sum = *s.Count
select {
case <-t.C:
}
//log.NewWarn("", " system stat ", s.ModuleName, s.PrintArgs, *s.Count-sum, "total:", *s.Count)
log.NewWarn("", " system stat ", s.ModuleName, s.PrintArgs, *s.Count-sum, "total:", *s.Count)
}
}

Write Preview
Loading…
Cancel
Save