Merge branch 'main' into patch-1

pull/167/head
Gordon 4 years ago committed by GitHub
commit 96cf3aadd8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -1 +1 @@
Subproject commit 1c6c7af5393b3e9eefbaf16b673519ca863a6c2c
Subproject commit dd23b4b185a273d38e73d51990c531e37ab5ba3b

@ -59,13 +59,13 @@ func main() {
groupRouterGroup.POST("/transfer_group", group.TransferGroupOwner) //1
groupRouterGroup.POST("/get_recv_group_applicationList", group.GetRecvGroupApplicationList) //1
groupRouterGroup.POST("/get_user_req_group_applicationList", group.GetUserReqGroupApplicationList)
groupRouterGroup.POST("/get_groups_info", group.GetGroupsInfo) //1
groupRouterGroup.POST("/kick_group", group.KickGroupMember) //1
groupRouterGroup.POST("/get_group_member_list", group.GetGroupMemberList) //no use
groupRouterGroup.POST("/get_group_all_member_list", group.GetGroupAllMemberList) //1
groupRouterGroup.POST("/get_group_members_info", group.GetGroupMembersInfo) //1
groupRouterGroup.POST("/invite_user_to_group", group.InviteUserToGroup) //1
groupRouterGroup.POST("/get_joined_group_list", group.GetJoinedGroupList) //1
groupRouterGroup.POST("/get_groups_info", group.GetGroupsInfo) //1
groupRouterGroup.POST("/kick_group", group.KickGroupMember) //1
groupRouterGroup.POST("/get_group_member_list", group.GetGroupMemberList) //no use
groupRouterGroup.POST("/get_group_all_member_list", group.GetGroupAllMemberList) //1
groupRouterGroup.POST("/get_group_members_info", group.GetGroupMembersInfo) //1
groupRouterGroup.POST("/invite_user_to_group", group.InviteUserToGroup) //1
groupRouterGroup.POST("/get_joined_group_list", group.GetJoinedGroupList) //1
}
//certificate
authRouterGroup := r.Group("/auth")
@ -77,6 +77,7 @@ func main() {
thirdGroup := r.Group("/third")
{
thirdGroup.POST("/tencent_cloud_storage_credential", apiThird.TencentCloudStorageCredential)
thirdGroup.POST("/ali_oss_credential", apiThird.AliOSSCredential)
thirdGroup.POST("/minio_storage_credential", apiThird.MinioStorageCredential)
}
//Message

@ -19,6 +19,7 @@ mysql:
dbMaxLifeTime: 120
mongo:
dbUri: ""#当dbUri值不为空则直接使用该值
dbAddress: [ 127.0.0.1:37017 ] #mongo地址 目前仅支持单机,默认即可
dbDirect: false
dbTimeout: 10
@ -93,6 +94,16 @@ credential: #腾讯cos发送图片、视频、文件时需要请自行申
endpoint: http://127.0.0.1:9000
accessKeyID: user12345
secretAccessKey: key12345
ali: # ali oss
regionID: "oss-cn-beijing"
accessKeyID: ""
accessKeySecret: ""
stsEndpoint: "sts.cn-beijing.aliyun.com"
ossEndpoint: "oss-cn-beijing.aliyuncs.com"
bucket: "bucket1"
finalHost: "http://bucket1.oss-cn-beijing.aliyuncs.com"
stsDurationSeconds: 3600
OssRoleArn: "acs:ram::xxx:role/xxx"
rpcport: #rpc服务端口 默认即可
@ -178,7 +189,8 @@ tokenpolicy:
accessSecret: "open_im_server" #token生成相关默认即可
# Token effective time day as a unit
accessExpire: 3650 #token过期时间 默认即可
messageverify:
friendVerify: false
# c2c:
# callbackBeforeSendMsg:
# switch: false

@ -19,6 +19,7 @@ mysql:
dbMaxLifeTime: 120
mongo:
dbUri: ""#当dbUri值不为空则直接使用该值
dbAddress: [ openim_mongo:27017 ]
dbDirect: false
dbTimeout: 10

@ -93,7 +93,7 @@ services:
command: /usr/local/bin/etcd --name etcd0 --data-dir /etcd-data --listen-client-urls http://0.0.0.0:2379 --advertise-client-urls http://0.0.0.0:2379 --listen-peer-urls http://0.0.0.0:2380 --initial-advertise-peer-urls http://0.0.0.0:2380 --initial-cluster etcd0=http://0.0.0.0:2380 --initial-cluster-token tkn --initial-cluster-state new
open_im_server:
image: openim/open_im_server:v2.0.2
image: openim/open_im_server:v2.0.4
container_name: open_im_server
volumes:
- ./logs:/Open-IM-Server/logs

@ -7,6 +7,7 @@ require (
github.com/Shopify/toxiproxy v2.1.4+incompatible // indirect
github.com/alibabacloud-go/darabonba-openapi v0.1.11
github.com/alibabacloud-go/dysmsapi-20170525/v2 v2.0.8
github.com/alibabacloud-go/sts-20150401 v1.1.0
github.com/alibabacloud-go/tea v1.1.17
github.com/antonfisher/nested-logrus-formatter v1.3.0
github.com/bwmarrin/snowflake v0.3.0
@ -52,6 +53,7 @@ require (
google.golang.org/protobuf v1.27.1
gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
gopkg.in/ini.v1 v1.66.2 // indirect
gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
sigs.k8s.io/yaml v1.2.0 // indirect

@ -50,6 +50,7 @@ github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWso
github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/alibabacloud-go/darabonba-openapi v0.1.7/go.mod h1:6FV1Bt1AItYIlC2rVopPTumrRNtkfPBmrPVAZ8v2bLk=
github.com/alibabacloud-go/darabonba-openapi v0.1.11 h1:w59gtSA0s87p0U5NNG/N/PIHsRP3rtj7qCP9hx9+GL8=
github.com/alibabacloud-go/darabonba-openapi v0.1.11/go.mod h1:MPJMxv7HYrFm5m9uOZWkDYsAWyZztEgnBRfk9Fg0eIU=
github.com/alibabacloud-go/darabonba-string v1.0.0/go.mod h1:93cTfV3vuPhhEwGGpKKqhVW4jLe7tDpo3LUM0i0g6mA=
@ -59,8 +60,11 @@ github.com/alibabacloud-go/dysmsapi-20170525/v2 v2.0.8 h1:KXMiCg99Jx7B6V+DlRFbWw
github.com/alibabacloud-go/dysmsapi-20170525/v2 v2.0.8/go.mod h1:8aL6tSyQIWJygF7W/Vqxdf/QDbN2S+u57k36bEA8hD8=
github.com/alibabacloud-go/endpoint-util v1.1.0 h1:r/4D3VSw888XGaeNpP994zDUaxdgTSHBbVfZlzf6b5Q=
github.com/alibabacloud-go/endpoint-util v1.1.0/go.mod h1:O5FuCALmCKs2Ff7JFJMudHs0I5EBgecXXxZRyswlEjE=
github.com/alibabacloud-go/openapi-util v0.0.8/go.mod h1:sQuElr4ywwFRlCCberQwKRFhRzIyG4QTP/P4y1CJ6Ws=
github.com/alibabacloud-go/openapi-util v0.0.9 h1:Z0DP4LFzkM/rW2nxOMiiLoQVZSeE3jVc5jrZ9Fd/UX0=
github.com/alibabacloud-go/openapi-util v0.0.9/go.mod h1:sQuElr4ywwFRlCCberQwKRFhRzIyG4QTP/P4y1CJ6Ws=
github.com/alibabacloud-go/sts-20150401 v1.1.0 h1:1yVyKz02ES6aKo3xVjmoPLBH1OAmmSqPkhKRdjEkmYs=
github.com/alibabacloud-go/sts-20150401 v1.1.0/go.mod h1:QW4O/c7Hp4krHYt+6xwnoG8EyZW3V9GYkl6EgIBmxJc=
github.com/alibabacloud-go/tea v1.1.0/go.mod h1:IkGyUSX4Ba1V+k4pCtJUc6jDpZLFph9QMy2VUPTwukg=
github.com/alibabacloud-go/tea v1.1.7/go.mod h1:/tmnEaQMyb4Ky1/5D+SE1BAsa5zj/KeGOFfwYm3N/p4=
github.com/alibabacloud-go/tea v1.1.8/go.mod h1:/tmnEaQMyb4Ky1/5D+SE1BAsa5zj/KeGOFfwYm3N/p4=
@ -904,8 +908,9 @@ gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df h1:n7WqCuqOuCbNr617RXOY0AW
gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df/go.mod h1:LRQQ+SO6ZHR7tOkpBDuZnXENFzX8qRjMDMyPD6BRkCw=
gopkg.in/ini.v1 v1.56.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
gopkg.in/ini.v1 v1.57.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
gopkg.in/ini.v1 v1.63.2 h1:tGK/CyBg7SMzb60vP1M03vNZ3VDu3wGQJwn7Sxi9r3c=
gopkg.in/ini.v1 v1.63.2/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
gopkg.in/ini.v1 v1.66.2 h1:XfR1dOYubytKy4Shzc2LHrrGhU0lDCfDGG1yLPmpgsI=
gopkg.in/ini.v1 v1.66.2/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 h1:VpOs+IwYnYBaFnrNAeB8UUWtL3vEUnzSCL1nVjPhqrw=
gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22/go.mod h1:yeKp02qBN3iKW1OzL3MGk2IdtZzaj7SFntXj72NppTA=
gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=

@ -0,0 +1,95 @@
package apiThird
import (
api "Open_IM/pkg/base_info"
"Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log"
"Open_IM/pkg/common/token_verify"
"fmt"
openapi "github.com/alibabacloud-go/darabonba-openapi/client"
sts20150401 "github.com/alibabacloud-go/sts-20150401/client"
"github.com/alibabacloud-go/tea/tea"
"github.com/fatih/structs"
//"github.com/fatih/structs"
"github.com/gin-gonic/gin"
"net/http"
"time"
)
var stsClient *sts20150401.Client
/**
* 使AK&SKClient
* @param accessKeyId
* @param accessKeySecret
* @return Client
* @throws Exception
*/
func getStsClient() *sts20150401.Client {
if stsClient != nil {
return stsClient
}
conf := &openapi.Config{
// 您的AccessKey ID
AccessKeyId: tea.String(config.Config.Credential.Ali.AccessKeyID),
// 您的AccessKey Secret
AccessKeySecret: tea.String(config.Config.Credential.Ali.AccessKeySecret),
// Endpoint
Endpoint: tea.String(config.Config.Credential.Ali.StsEndpoint),
}
result, err := sts20150401.NewClient(conf)
if err != nil {
log.NewError("", "alists client初始化失败 ", err)
}
stsClient = result
return stsClient
}
func AliOSSCredential(c *gin.Context) {
req := api.OSSCredentialReq{}
if err := c.BindJSON(&req); err != nil {
log.NewError("0", "BindJSON failed ", err.Error())
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
ok, userID := token_verify.GetUserIDFromToken(c.Request.Header.Get("token"))
if !ok {
log.NewError(req.OperationID, "GetUserIDFromToken false ", c.Request.Header.Get("token"))
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": "GetUserIDFromToken failed"})
return
}
log.NewInfo(req.OperationID, "AliOSSCredential args ", userID)
stsResp, err := getStsClient().AssumeRole(&sts20150401.AssumeRoleRequest{
DurationSeconds: tea.Int64(config.Config.Credential.Ali.StsDurationSeconds),
Policy: nil,
RoleArn: tea.String(config.Config.Credential.Ali.OssRoleArn),
RoleSessionName: tea.String(fmt.Sprintf("%s-%d", userID, time.Now().Unix())),
})
resp := api.OSSCredentialResp{}
if err != nil {
resp.ErrCode = constant.ErrTencentCredential.ErrCode
resp.ErrMsg = err.Error()
} else {
resp = api.OSSCredentialResp{
CommResp: api.CommResp{},
OssData: api.OSSCredentialRespData{
Endpoint: config.Config.Credential.Ali.OssEndpoint,
AccessKeyId: *stsResp.Body.Credentials.AccessKeyId,
AccessKeySecret: *stsResp.Body.Credentials.AccessKeySecret,
Token: *stsResp.Body.Credentials.SecurityToken,
Bucket: config.Config.Credential.Ali.Bucket,
FinalHost: config.Config.Credential.Ali.FinalHost,
},
Data: nil,
}
}
resp.Data = structs.Map(&resp.OssData)
log.NewInfo(req.OperationID, "AliOSSCredential return ", resp)
c.JSON(http.StatusOK, resp)
}

@ -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}

@ -128,7 +128,7 @@ func GetGroupByName(c *gin.Context) {
func CreateGroup(c *gin.Context) {
var (
req cms_api_struct.CreateGroupRequest
_ cms_api_struct.CreateGroupResponse
_ cms_api_struct.CreateGroupResponse
reqPb pbGroup.CreateGroupReq
)
if err := c.BindJSON(&req); err != nil {
@ -253,9 +253,9 @@ func OpenPrivateChat(c *gin.Context) {
func GetGroupMembers(c *gin.Context) {
var (
req cms_api_struct.GetGroupMembersRequest
req cms_api_struct.GetGroupMembersRequest
reqPb pbGroup.GetGroupMembersCMSReq
resp cms_api_struct.GetGroupMembersResponse
resp cms_api_struct.GetGroupMembersResponse
)
if err := c.ShouldBindQuery(&req); err != nil {
log.NewError(reqPb.OperationID, utils.GetSelfFuncName(), "ShouldBindQuery failed ", err.Error())
@ -286,21 +286,20 @@ 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
resp cms_api_struct.RemoveGroupMembersResponse
req cms_api_struct.RemoveGroupMembersRequest
resp cms_api_struct.RemoveGroupMembersResponse
reqPb pbGroup.AddGroupMembersCMSReq
)
if err := c.BindJSON(&req); err != nil {
log.NewError(reqPb.OperationId, utils.GetSelfFuncName(),"BindJSON failed ", err.Error())
log.NewError(reqPb.OperationId, utils.GetSelfFuncName(), "BindJSON failed ", err.Error())
openIMHttp.RespHttp200(c, constant.ErrArgs, nil)
return
}
@ -321,12 +320,12 @@ func AddGroupMembers(c *gin.Context) {
func RemoveGroupMembers(c *gin.Context) {
var (
req cms_api_struct.RemoveGroupMembersRequest
resp cms_api_struct.RemoveGroupMembersResponse
req cms_api_struct.RemoveGroupMembersRequest
resp cms_api_struct.RemoveGroupMembersResponse
reqPb pbGroup.RemoveGroupMembersCMSReq
)
if err := c.BindJSON(&req); err != nil {
log.NewError(reqPb.OperationID, utils.GetSelfFuncName(),"BindJSON failed ", err.Error())
log.NewError(reqPb.OperationID, utils.GetSelfFuncName(), "BindJSON failed ", err.Error())
openIMHttp.RespHttp200(c, constant.ErrArgs, nil)
return
}
@ -347,12 +346,12 @@ func RemoveGroupMembers(c *gin.Context) {
func DeleteGroup(c *gin.Context) {
var (
req cms_api_struct.DeleteGroupRequest
_ cms_api_struct.DeleteGroupResponse
req cms_api_struct.DeleteGroupRequest
_ cms_api_struct.DeleteGroupResponse
reqPb pbGroup.DeleteGroupReq
)
if err := c.BindJSON(&req); err != nil {
log.NewError(reqPb.OperationID, utils.GetSelfFuncName(),"BindJSON failed ", err.Error())
log.NewError(reqPb.OperationID, utils.GetSelfFuncName(), "BindJSON failed ", err.Error())
openIMHttp.RespHttp200(c, constant.ErrArgs, nil)
return
}
@ -370,12 +369,12 @@ func DeleteGroup(c *gin.Context) {
func SetGroupMaster(c *gin.Context) {
var (
req cms_api_struct.SetGroupMasterRequest
_ cms_api_struct.SetGroupMasterResponse
req cms_api_struct.SetGroupMasterRequest
_ cms_api_struct.SetGroupMasterResponse
reqPb pbGroup.OperateUserRoleReq
)
if err := c.BindJSON(&req); err != nil {
log.NewError(reqPb.OperationID, utils.GetSelfFuncName(),"BindJSON failed ", err.Error())
log.NewError(reqPb.OperationID, utils.GetSelfFuncName(), "BindJSON failed ", err.Error())
openIMHttp.RespHttp200(c, constant.ErrArgs, nil)
return
}
@ -395,8 +394,8 @@ func SetGroupMaster(c *gin.Context) {
func SetGroupOrdinaryUsers(c *gin.Context) {
var (
req cms_api_struct.SetGroupMemberRequest
_ cms_api_struct.AdminLoginResponse
req cms_api_struct.SetGroupMemberRequest
_ cms_api_struct.AdminLoginResponse
reqPb pbGroup.OperateUserRoleReq
)
if err := c.BindJSON(&req); err != nil {
@ -420,8 +419,8 @@ func SetGroupOrdinaryUsers(c *gin.Context) {
func AlterGroupInfo(c *gin.Context) {
var (
req cms_api_struct.AlterGroupInfoRequest
_ cms_api_struct.SetGroupMasterResponse
req cms_api_struct.AlterGroupInfoRequest
_ cms_api_struct.SetGroupMasterResponse
reqPb pbGroup.SetGroupInfoReq
)
if err := c.BindJSON(&req); err != nil {
@ -431,12 +430,12 @@ func AlterGroupInfo(c *gin.Context) {
}
reqPb.OpUserID = c.MustGet("userID").(string)
reqPb.GroupInfo = &commonPb.GroupInfo{
GroupID: req.GroupID,
GroupName: req.GroupName,
Introduction: req.Introduction,
Notification: req.Notification,
FaceURL: req.ProfilePhoto,
GroupType: int32(req.GroupType),
GroupID: req.GroupID,
GroupName: req.GroupName,
Introduction: req.Introduction,
Notification: req.Notification,
FaceURL: req.ProfilePhoto,
GroupType: int32(req.GroupType),
}
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImGroupName)
client := pbGroup.NewGroupClient(etcdConn)

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

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

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

@ -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

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

@ -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)
}
}
@ -647,9 +647,9 @@ func (s *groupServer) TransferGroupOwner(_ context.Context, req *pbGroup.Transfe
}
func (s *groupServer) GetGroupById(_ context.Context, req *pbGroup.GetGroupByIdReq) (*pbGroup.GetGroupByIdResp, error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp := &pbGroup.GetGroupByIdResp{CMSGroup: &pbGroup.CMSGroup{
GroupInfo: &open_im_sdk.GroupInfo{},
GroupInfo: &open_im_sdk.GroupInfo{},
}}
group, err := imdb.GetGroupById(req.GroupId)
if err != nil {
@ -687,7 +687,7 @@ func (s *groupServer) GetGroup(_ context.Context, req *pbGroup.GetGroupReq) (*pb
log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetGroupsByName error", req.String())
return resp, http.WrapError(constant.ErrDB)
}
nums, err := imdb.GetGroupsCountNum(db.Group{GroupName:req.GroupName})
nums, err := imdb.GetGroupsCountNum(db.Group{GroupName: req.GroupName})
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetGroupsCountNum error", err.Error())
return resp, http.WrapError(constant.ErrDB)
@ -712,7 +712,7 @@ func (s *groupServer) GetGroup(_ context.Context, req *pbGroup.GetGroupReq) (*pb
CreatorUserID: v.CreatorUserID,
},
GroupMasterName: groupMember.Nickname,
GroupMasterId: groupMember.UserID,
GroupMasterId: groupMember.UserID,
})
}
return resp, nil
@ -753,7 +753,7 @@ func (s *groupServer) GetGroups(_ context.Context, req *pbGroup.GetGroupsReq) (*
CreatorUserID: v.CreatorUserID,
CreateTime: uint32(v.CreateTime.Unix()),
},
GroupMasterId: groupMember.UserID,
GroupMasterId: groupMember.UserID,
GroupMasterName: groupMember.Nickname,
})
}
@ -808,11 +808,11 @@ func (s *groupServer) OperateUserRole(_ context.Context, req *pbGroup.OperateUse
}
func (s *groupServer) GetGroupMembersCMS(_ context.Context, req *pbGroup.GetGroupMembersCMSReq) (*pbGroup.GetGroupMembersCMSResp, error) {
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "args:", req.String())
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "args:", req.String())
resp := &pbGroup.GetGroupMembersCMSResp{}
groupMembers, err := imdb.GetGroupMembersByGroupIdCMS(req.GroupId, req.UserName, req.Pagination.ShowNumber, req.Pagination.PageNumber)
if err != nil {
log.NewError(req.OperationID, utils.GetSelfFuncName(),"GetGroupMembersByGroupIdCMS Error", err.Error())
log.NewError(req.OperationID, utils.GetSelfFuncName(), "GetGroupMembersByGroupIdCMS Error", err.Error())
return resp, http.WrapError(constant.ErrDB)
}
groupMembersCount, err := imdb.GetGroupMembersCount(req.GroupId, req.UserName)
@ -824,13 +824,13 @@ func (s *groupServer) GetGroupMembersCMS(_ context.Context, req *pbGroup.GetGrou
resp.MemberNums = groupMembersCount
for _, groupMember := range groupMembers {
resp.Members = append(resp.Members, &open_im_sdk.GroupMemberFullInfo{
GroupID: req.GroupId,
UserID: groupMember.UserID,
RoleLevel: groupMember.RoleLevel,
JoinTime: groupMember.JoinTime.Unix(),
Nickname: groupMember.Nickname,
FaceURL: groupMember.FaceURL,
JoinSource: groupMember.JoinSource,
GroupID: req.GroupId,
UserID: groupMember.UserID,
RoleLevel: groupMember.RoleLevel,
JoinTime: int32(groupMember.JoinTime.Unix()),
Nickname: groupMember.Nickname,
FaceURL: groupMember.FaceURL,
JoinSource: groupMember.JoinSource,
})
}
resp.Pagination = &open_im_sdk.ResponsePagination{
@ -894,7 +894,7 @@ func (s *groupServer) AddGroupMembersCMS(_ context.Context, req *pbGroup.AddGrou
if err := imdb.InsertIntoGroupMember(groupMember); err != nil {
log.NewError(req.OperationId, utils.GetSelfFuncName(), "InsertIntoGroupMember failed", req.String())
resp.Failed = append(resp.Failed, userId)
} else {
} else {
resp.Success = append(resp.Success, userId)
}
}
@ -902,9 +902,8 @@ 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())
log.NewInfo(req.OperationID, utils.GetSelfFuncName(), "req: ", req.String())
resp := &pbGroup.GetUserReqApplicationListResp{}
groupRequests, err := imdb.GetUserReqGroupByUserID(req.UserID)
if err != nil {

@ -21,7 +21,7 @@ type ImportFriendReq struct {
FromUserID string `json:"fromUserID" binding:"required"`
}
type UserIDResult struct {
UserID string `json:"userID""`
UserID string `json:"userID"`
Result int32 `json:"result"`
}
type ImportFriendResp struct {

@ -0,0 +1,22 @@
package base_info
type OSSCredentialReq struct {
OperationID string `json:"operationID"`
Filename string `json:"filename"`
FileType string `json:"file_type"`
}
type OSSCredentialRespData struct {
Endpoint string `json:"endpoint"`
AccessKeyId string `json:"access_key_id"`
AccessKeySecret string `json:"access_key_secret"`
Token string `json:"token"`
Bucket string `json:"bucket"`
FinalHost string `json:"final_host"`
}
type OSSCredentialResp struct {
CommResp
OssData OSSCredentialRespData `json:"-"`
Data map[string]interface{} `json:"data"`
}

@ -20,8 +20,8 @@ var (
var Config config
type callBackConfig struct {
Enable bool `yaml:"enable"`
CallbackTimeOut int `yaml:"callbackTimeOut"`
Enable bool `yaml:"enable"`
CallbackTimeOut int `yaml:"callbackTimeOut"`
CallbackFailedContinue bool `CallbackFailedContinue`
}
@ -31,7 +31,7 @@ type config struct {
Api struct {
GinPort []int `yaml:"openImApiPort"`
}
CmsApi struct{
CmsApi struct {
GinPort []int `yaml:"openImCmsApiPort"`
}
Sdk struct {
@ -45,6 +45,17 @@ type config struct {
SecretID string `yaml:"secretID"`
SecretKey string `yaml:"secretKey"`
}
Ali struct {
RegionID string `yaml:"regionID"`
AccessKeyID string `yaml:"accessKeyID"`
AccessKeySecret string `yaml:"accessKeySecret"`
StsEndpoint string `yaml:"stsEndpoint"`
OssEndpoint string `yaml:"ossEndpoint"`
Bucket string `yaml:"bucket"`
FinalHost string `yaml:"finalHost"`
StsDurationSeconds int64 `yaml:"stsDurationSeconds"`
OssRoleArn string `yaml:"OssRoleArn"`
}
Minio struct {
Bucket string `yaml:"bucket"`
Location string `yaml:"location"`
@ -66,6 +77,7 @@ type config struct {
DBMaxLifeTime int `yaml:"dbMaxLifeTime"`
}
Mongo struct {
DBUri string `yaml:"dbUri"` // 当dbUri值不为空则直接使用该值
DBAddress []string `yaml:"dbAddress"`
DBDirect bool `yaml:"dbDirect"`
DBTimeout int `yaml:"dbTimeout"`
@ -182,12 +194,12 @@ type config struct {
}
Callback struct {
CallbackUrl string `yaml:"callbackUrl"`
CallbackUrl string `yaml:"callbackUrl"`
CallbackBeforeSendSingleMsg callBackConfig `yaml:"callbackbeforeSendSingleMsg"`
CallbackAfterSendSingleMsg callBackConfig `yaml:"callbackAfterSendSingleMsg"`
CallbackBeforeSendGroupMsg callBackConfig `yaml:"callbackBeforeSendGroupMsg"`
CallbackAfterSendGroupMsg callBackConfig `yaml:"callbackAfterSendGroupMsg"`
CallbackWordFilter callBackConfig `yaml:"callbackWordFilter"`
CallbackAfterSendSingleMsg callBackConfig `yaml:"callbackAfterSendSingleMsg"`
CallbackBeforeSendGroupMsg callBackConfig `yaml:"callbackBeforeSendGroupMsg"`
CallbackAfterSendGroupMsg callBackConfig `yaml:"callbackAfterSendGroupMsg"`
CallbackWordFilter callBackConfig `yaml:"callbackWordFilter"`
} `yaml:"callback"`
Notification struct {
///////////////////////group/////////////////////////////

@ -20,8 +20,10 @@ const (
WSGetNewestSeq = 1001
WSPullMsgBySeqList = 1002
WSSendMsg = 1003
WSSendSignalMsg = 1004
WSPushMsg = 2001
WSKickOnlineMsg = 2002
WsLogoutMsg = 2003
WSDataError = 3001
///ContentType
@ -139,16 +141,16 @@ const (
//callbackCommand
CallbackBeforeSendSingleMsgCommand = "callbackBeforeSendSingleMsgCommand"
CallbackAfterSendSingleMsgCommand = "callbackAfterSendSingleMsgCommand"
CallbackBeforeSendGroupMsgCommand = "callbackBeforeSendGroupMsgCommand"
CallbackAfterSendGroupMsgCommand = "callbackAfterSendGroupMsgCommand"
CallbackWordFilterCommand = "callbackWordFilterCommand"
CallbackAfterSendSingleMsgCommand = "callbackAfterSendSingleMsgCommand"
CallbackBeforeSendGroupMsgCommand = "callbackBeforeSendGroupMsgCommand"
CallbackAfterSendGroupMsgCommand = "callbackAfterSendGroupMsgCommand"
CallbackWordFilterCommand = "callbackWordFilterCommand"
//callback actionCode
ActionAllow = 0
ActionAllow = 0
ActionForbidden = 1
//callback callbackHandleCode
CallbackHandleSuccess = 0
CallbackHandleFailed = 1
CallbackHandleFailed = 1
)
var ContentType2PushContent = map[int64]string{

@ -42,9 +42,14 @@ func init() {
// mongo init
// "mongodb://sysop:moon@localhost/records"
uri := "mongodb://sample.host:27017/?maxPoolSize=20&w=majority"
uri = fmt.Sprintf("mongodb://%s/%s/?maxPoolSize=%d",
config.Config.Mongo.DBAddress[0],config.Config.Mongo.DBDatabase,
config.Config.Mongo.DBMaxPoolSize)
if config.Config.Mongo.DBUri != "" {
// example: mongodb://$user:$password@mongo1.mongo:27017,mongo2.mongo:27017,mongo3.mongo:27017/$DBDatabase/?replicaSet=rs0&readPreference=secondary&authSource=admin&maxPoolSize=$DBMaxPoolSize
uri = config.Config.Mongo.DBUri
} else {
uri = fmt.Sprintf("mongodb://%s/%s/?maxPoolSize=%d",
config.Config.Mongo.DBAddress[0], config.Config.Mongo.DBDatabase,
config.Config.Mongo.DBMaxPoolSize)
}
mongoClient, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(uri))
if err != nil{

@ -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
}

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

File diff suppressed because it is too large Load Diff

@ -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 {
}

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

@ -52,13 +52,13 @@ else
fi
check=$(ps aux | grep -w ./${timer_task_name} | grep -v grep | wc -l)
if [ $check -ge 1 ]; then
echo -e ${GREEN_PREFIX}"none port has been listening,belongs service is openImMsgTimer"${COLOR_SUFFIX}
else
echo -e ${RED_PREFIX}"openImMsgTimer service does not start normally"${COLOR_SUFFIX}
echo -e ${RED_PREFIX}"please check ../logs/openIM.log "${COLOR_SUFFIX}
exit -1
fi
#check=$(ps aux | grep -w ./${timer_task_name} | grep -v grep | wc -l)
#if [ $check -ge 1 ]; then
# echo -e ${GREEN_PREFIX}"none port has been listening,belongs service is openImMsgTimer"${COLOR_SUFFIX}
#else
# echo -e ${RED_PREFIX}"openImMsgTimer service does not start normally"${COLOR_SUFFIX}
# echo -e ${RED_PREFIX}"please check ../logs/openIM.log "${COLOR_SUFFIX}
# exit -1
#fi
echo -e ${YELLOW_PREFIX}"all services launch success"${COLOR_SUFFIX}

Loading…
Cancel
Save