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

Merge remote-tracking branch 'origin/main'

Browse Source
pull/4/head
away 4 years ago
parent 7c695e4431 c96498e531
commit f2bbd4b924
16 changed files with 1233 additions and 0 deletions
Show Stats Download Patch File Download Diff File

26
src/api/Makefile
Unescape View File

@ -0,0 +1,26 @@
.PHONY: all build run gotool install clean help
BINARY_NAME=open_im_api
BIN_DIR=../../bin/
LAN_FILE=.go
GO_FILE:=${BINARY_NAME}${LAN_FILE}
all: gotool build
build:
CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o ${BINARY_NAME} ${GO_FILE}
run:
@go run ./
gotool:
go fmt ./
go vet ./
install:
make build
mv ${BINARY_NAME} ${BIN_DIR}
clean:
@if [ -f ${BINARY_NAME} ] ; then rm ${BINARY_NAME} ; fi

90
src/api/auth/user_register.go
Unescape View File

@ -0,0 +1,90 @@
package apiAuth
import (
"Open_IM/src/common/config"
"Open_IM/src/common/log"
pbAuth "Open_IM/src/proto/auth"
"context"
"github.com/gin-gonic/gin"
"github.com/skiffer-git/grpc-etcdv3/getcdv3"
"net/http"
"strings"
)
type paramsUserRegister struct {
Secret string `json:"secret" binding:"required,max=32"`
Platform int32 `json:"platform" binding:"required,min=1,max=7"`
UID string `json:"uid" binding:"required,min=1,max=64"`
Name string `json:"name" binding:"required,min=1,max=64"`
Icon string `json:"icon" binding:"omitempty,max=1024"`
Gender int32 `json:"gender" binding:"omitempty,oneof=0 1 2"`
Mobile string `json:"mobile" binding:"omitempty,max=32"`
Birth string `json:"birth" binding:"omitempty,max=16"`
Email string `json:"email" binding:"omitempty,max=64"`
Ex string `json:"ex" binding:"omitempty,max=1024"`
}
func newUserRegisterReq(params *paramsUserRegister) *pbAuth.UserRegisterReq {
pbData := pbAuth.UserRegisterReq{
UID: params.UID,
Name: params.Name,
Icon: params.Icon,
Gender: params.Gender,
Mobile: params.Mobile,
Birth: params.Birth,
Email: params.Email,
Ex: params.Ex,
}
return &pbData
}
func UserRegister(c *gin.Context) {
log.Info("", "", "api user_register init ....")
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.RpcGetTokenName)
client := pbAuth.NewAuthClient(etcdConn)
params := paramsUserRegister{}
if err := c.BindJSON(&params); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
pbData := newUserRegisterReq(&params)
log.Info("", "", "api user_register is server, [data: %s]", pbData.String())
reply, err := client.UserRegister(context.Background(), pbData)
if err != nil || !reply.Success {
log.Error("", "", "api user_register call rpc fail, [data: %s] [err: %s]", pbData.String(), err.Error())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
return
}
log.Info("", "", "api user_register call rpc success, [data: %s] [reply: %s]", pbData.String(), reply.String())
pbDataToken := &pbAuth.UserTokenReq{
Platform: params.Platform,
UID: params.UID,
}
replyToken, err := client.UserToken(context.Background(), pbDataToken)
if err != nil {
log.Error("", "", "api user_register call rpc fail, [data: %s] [err: %s]", pbData.String(), err.Error())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
return
}
log.Info("", "", "api user_register call success, [data: %s] [reply: %s]", pbData.String(), reply.String())
if replyToken.ErrCode == 0 {
c.JSON(http.StatusOK, gin.H{
"errCode": replyToken.ErrCode,
"errMsg": replyToken.ErrMsg,
"data": gin.H{
"uid": pbData.UID,
"token": replyToken.Token,
"expiredTime": replyToken.ExpiredTime,
},
})
} else {
c.JSON(http.StatusOK, gin.H{
"errCode": replyToken.ErrCode,
"errMsg": replyToken.ErrMsg,
})
}
}

66
src/api/auth/user_token.go
Unescape View File

@ -0,0 +1,66 @@
package apiAuth
import (
"Open_IM/src/common/config"
"Open_IM/src/common/log"
pbAuth "Open_IM/src/proto/auth"
"context"
"github.com/gin-gonic/gin"
"github.com/skiffer-git/grpc-etcdv3/getcdv3"
"net/http"
"strings"
)
type paramsUserToken struct {
Secret string `json:"secret" binding:"required,max=32"`
Platform int32 `json:"platform" binding:"required,min=1,max=7"`
UID string `json:"uid" binding:"required,min=1,max=64"`
}
func newUserTokenReq(params *paramsUserToken) *pbAuth.UserTokenReq {
pbData := pbAuth.UserTokenReq{
Platform: params.Platform,
UID: params.UID,
}
return &pbData
}
func UserToken(c *gin.Context) {
log.Info("", "", "api user_token init ....")
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.RpcGetTokenName)
client := pbAuth.NewAuthClient(etcdConn)
params := paramsUserToken{}
if err := c.BindJSON(&params); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
pbData := newUserTokenReq(&params)
log.Info("", "", "api user_token is server, [data: %s]", pbData.String())
reply, err := client.UserToken(context.Background(), pbData)
if err != nil {
log.Error("", "", "api user_token call rpc fail, [data: %s] [err: %s]", pbData.String(), err.Error())
c.JSON(http.StatusInternalServerError, gin.H{"errCode": 500, "errMsg": err.Error()})
return
}
log.Info("", "", "api user_token call rpc success, [data: %s] [reply: %s]", pbData.String(), reply.String())
if reply.ErrCode == 0 {
c.JSON(http.StatusOK, gin.H{
"errCode": reply.ErrCode,
"errMsg": reply.ErrMsg,
"data": gin.H{
"uid": pbData.UID,
"token": reply.Token,
"expiredTime": reply.ExpiredTime,
},
})
} else {
c.JSON(http.StatusOK, gin.H{
"errCode": reply.ErrCode,
"errMsg": reply.ErrMsg,
})
}
}

59
src/api/chat/newest_seq.go
Unescape View File

@ -0,0 +1,59 @@
package apiChat
import (
"Open_IM/src/common/config"
"Open_IM/src/common/log"
pbMsg "Open_IM/src/proto/chat"
"Open_IM/src/utils"
"context"
"github.com/gin-gonic/gin"
"github.com/skiffer-git/grpc-etcdv3/getcdv3"
"net/http"
"strings"
)
type paramsUserNewestSeq struct {
ReqIdentifier int `json:"reqIdentifier" binding:"required"`
SendID string `json:"sendID" binding:"required"`
OperationID string `json:"operationID" binding:"required"`
MsgIncr int `json:"msgIncr" binding:"required"`
}
func UserNewestSeq(c *gin.Context) {
params := paramsUserNewestSeq{}
if err := c.BindJSON(&params); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
token := c.Request.Header.Get("token")
if !utils.VerifyToken(token, params.SendID) {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": "token validate err"})
return
}
pbData := pbMsg.GetNewSeqReq{}
pbData.UserID = params.SendID
pbData.OperationID = params.OperationID
grpcConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
if grpcConn == nil {
log.ErrorByKv("get grpcConn err", pbData.OperationID, "args", params)
}
msgClient := pbMsg.NewChatClient(grpcConn)
reply, err := msgClient.GetNewSeq(context.Background(), &pbData)
if err != nil {
log.ErrorByKv("rpc call failed to getNewSeq", pbData.OperationID, "err", err, "pbData", pbData.String())
return
}
c.JSON(http.StatusOK, gin.H{
"errCode": reply.ErrCode,
"errMsg": reply.ErrMsg,
"msgIncr": params.MsgIncr,
"reqIdentifier": params.ReqIdentifier,
"data": gin.H{
"seq": reply.Seq,
},
})
}

75
src/api/chat/pull_msg.go
Unescape View File

@ -0,0 +1,75 @@
package apiChat
import (
"Open_IM/src/common/config"
"Open_IM/src/common/log"
"Open_IM/src/proto/chat"
"Open_IM/src/utils"
"context"
"github.com/gin-gonic/gin"
"github.com/skiffer-git/grpc-etcdv3/getcdv3"
"net/http"
"strings"
)
type paramsUserPullMsg struct {
ReqIdentifier int `json:"reqIdentifier" binding:"required"`
SendID string `json:"sendID" binding:"required"`
OperationID string `json:"operationID" binding:"required"`
MsgIncr int `json:"msgIncr" binding:"required"`
Data struct {
SeqBegin int64 `json:"seqBegin" binding:"required"`
SeqEnd int64 `json:"seqEnd" binding:"required"`
}
}
func UserPullMsg(c *gin.Context) {
params := paramsUserPullMsg{}
if err := c.BindJSON(&params); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
return
}
token := c.Request.Header.Get("token")
if !utils.VerifyToken(token, params.SendID) {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": "token validate err"})
return
}
pbData := pbChat.PullMessageReq{}
pbData.UserID = params.SendID
pbData.OperationID = params.OperationID
pbData.SeqBegin = params.Data.SeqBegin
pbData.SeqEnd = params.Data.SeqEnd
grpcConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
msgClient := pbChat.NewChatClient(grpcConn)
reply, err := msgClient.PullMessage(context.Background(), &pbData)
if err != nil {
log.ErrorByKv("PullMessage error", pbData.OperationID, "err", err.Error())
return
}
log.InfoByKv("rpc call success to pullMsgRep", pbData.OperationID, "ReplyArgs", reply.String(), "maxSeq", reply.GetMaxSeq(),
"MinSeq", reply.GetMinSeq(), "singLen", len(reply.GetSingleUserMsg()), "groupLen", len(reply.GetGroupUserMsg()))
msg := make(map[string]interface{})
if v := reply.GetSingleUserMsg(); v != nil {
msg["single"] = v
} else {
msg["single"] = []pbChat.GatherFormat{}
}
if v := reply.GetGroupUserMsg(); v != nil {
msg["group"] = v
} else {
msg["group"] = []pbChat.GatherFormat{}
}
msg["maxSeq"] = reply.GetMaxSeq()
msg["minSeq"] = reply.GetMinSeq()
c.JSON(http.StatusOK, gin.H{
"errCode": reply.ErrCode,
"errMsg": reply.ErrMsg,
"msgIncr": params.MsgIncr,
"reqIdentifier": params.ReqIdentifier,
"data": msg,
})
}

96
src/api/chat/send_msg.go
Unescape View File

@ -0,0 +1,96 @@
package apiChat
import (
"Open_IM/src/common/config"
"Open_IM/src/common/log"
pbChat "Open_IM/src/proto/chat"
"Open_IM/src/utils"
"context"
"github.com/gin-gonic/gin"
"github.com/skiffer-git/grpc-etcdv3/getcdv3"
"net/http"
"strings"
)
type paramsUserSendMsg struct {
ReqIdentifier int32 `json:"reqIdentifier" binding:"required"`
PlatformID int32 `json:"platformID" binding:"required"`
SendID string `json:"sendID" binding:"required"`
OperationID string `json:"operationID" binding:"required"`
MsgIncr int32 `json:"msgIncr" binding:"required"`
Data struct {
SessionType int32 `json:"sessionType" binding:"required"`
MsgFrom int32 `json:"msgFrom" binding:"required"`
ContentType int32 `json:"contentType" binding:"required"`
RecvID string `json:"recvID" binding:"required"`
ForceList []string `json:"forceList" binding:"required"`
Content string `json:"content" binding:"required"`
Options map[string]interface{} `json:"options" binding:"required"`
ClientMsgID string `json:"clientMsgID" binding:"required"`
OffLineInfo map[string]interface{} `json:"offlineInfo" binding:"required"`
Ex map[string]interface{} `json:"ext"`
}
}
func newUserSendMsgReq(token string, params *paramsUserSendMsg) *pbChat.UserSendMsgReq {
pbData := pbChat.UserSendMsgReq{
ReqIdentifier: params.ReqIdentifier,
Token: token,
SendID: params.SendID,
OperationID: params.OperationID,
MsgIncr: params.MsgIncr,
PlatformID: params.PlatformID,
SessionType: params.Data.SessionType,
MsgFrom: params.Data.MsgFrom,
ContentType: params.Data.ContentType,
RecvID: params.Data.RecvID,
ForceList: params.Data.ForceList,
Content: params.Data.Content,
Options: utils.MapToJsonString(params.Data.Options),
ClientMsgID: params.Data.ClientMsgID,
OffLineInfo: utils.MapToJsonString(params.Data.OffLineInfo),
Ex: utils.MapToJsonString(params.Data.Ex),
}
return &pbData
}
func UserSendMsg(c *gin.Context) {
params := paramsUserSendMsg{}
if err := c.BindJSON(&params); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": err.Error()})
log.ErrorByKv("json unmarshal err", "", "err", err.Error(), "data", c.PostForm("data"))
return
}
token := c.Request.Header.Get("token")
if !utils.VerifyToken(token, params.SendID) {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": "token validate err"})
return
}
log.InfoByKv("Ws call success to sendMsgReq", params.OperationID, "Parameters", params)
pbData := newUserSendMsgReq(token, &params)
log.Info("", "", "api UserSendMsg call start..., [data: %s]", pbData.String())
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOfflineMessageName)
client := pbChat.NewChatClient(etcdConn)
log.Info("", "", "api UserSendMsg call, api call rpc...")
reply, _ := client.UserSendMsg(context.Background(), pbData)
log.Info("", "", "api UserSendMsg call end..., [data: %s] [reply: %s]", pbData.String(), reply.String())
c.JSON(http.StatusOK, gin.H{
"errCode": 0,
"errMsg": "",
"msgIncr": reply.MsgIncr,
"reqIdentifier": reply.ReqIdentifier,
"data": gin.H{
"clientMsgID": reply.ClientMsgID,
"serverMsgID": reply.ServerMsgID,
},
})
}

75
src/api/open_im_api.go
Unescape View File

@ -0,0 +1,75 @@
package main
import (
apiAuth "Open_IM/src/api/auth"
apiChat "Open_IM/src/api/chat"
"Open_IM/src/api/friend"
apiThird "Open_IM/src/api/third"
"Open_IM/src/api/user"
"Open_IM/src/common/log"
"Open_IM/src/utils"
"flag"
"github.com/gin-gonic/gin"
"strconv"
)
func main() {
log.Info("", "", "api server running...")
r := gin.Default()
r.Use(utils.CorsHandler())
// user routing group, which handles user registration and login services
userRouterGroup := r.Group("/user")
{
userRouterGroup.POST("/update_user_info", user.UpdateUserInfo)
userRouterGroup.POST("/get_user_info", user.GetUserInfo)
}
//friend routing group
friendRouterGroup := r.Group("/friend")
{
friendRouterGroup.POST("/search_friend", friend.SearchFriend)
friendRouterGroup.POST("/add_friend", friend.AddFriend)
friendRouterGroup.POST("/get_friend_apply_list", friend.GetFriendApplyList)
friendRouterGroup.POST("/get_friend_list", friend.GetFriendList)
friendRouterGroup.POST("/add_blacklist", friend.AddBlacklist)
friendRouterGroup.POST("/get_blacklist", friend.GetBlacklist)
friendRouterGroup.POST("/remove_blacklist", friend.RemoveBlacklist)
friendRouterGroup.POST("/delete_friend", friend.DeleteFriend)
friendRouterGroup.POST("/add_friend_response", friend.AddFriendResponse)
friendRouterGroup.POST("/set_friend_comment", friend.SetFriendComment)
}
//group related routing group
/*groupRouterGroup := r.Group("/group")
{
groupRouterGroup.POST("/create_group", group.CreateGroup)
groupRouterGroup.POST("/get_group_list", group.GetGroupList)
groupRouterGroup.POST("/get_group_info", group.GetGroupInfo)
groupRouterGroup.POST("/delete_group_member", group.DeleteGroupMember)
groupRouterGroup.POST("/set_group_name", group.SetGroupName)
groupRouterGroup.POST("/set_group_bulletin", group.SetGroupBulletin)
groupRouterGroup.POST("/set_owner_group_nickname", group.SetOwnerGroupNickname)
groupRouterGroup.POST("/set_group_head_image", group.SetGroupHeadImage)
groupRouterGroup.POST("/member_exit_group", group.MemberExitGroup)
}*/
//certificate
authRouterGroup := r.Group("/auth")
{
authRouterGroup.POST("/user_register", apiAuth.UserRegister)
authRouterGroup.POST("/user_token", apiAuth.UserToken)
}
//Third service
thirdGroup := r.Group("/third")
{
thirdGroup.POST("/tencent_cloud_storage_credential", apiThird.TencentCloudStorageCredential)
}
//Message
chatGroup := r.Group("/chat")
{
chatGroup.POST("/newest_seq", apiChat.UserNewestSeq)
chatGroup.POST("/pull_msg", apiChat.UserPullMsg)
chatGroup.POST("/send_msg", apiChat.UserSendMsg)
}
ginPort := flag.Int("port", 10000, "get ginServerPort from cmd,default 10000 as port")
flag.Parse()
r.Run(utils.ServerIP + ":" + strconv.Itoa(*ginPort))
}

70
src/api/third/tencent_cloud_storage_credential.go
Unescape View File

@ -0,0 +1,70 @@
package apiThird
import (
"Open_IM/src/common/config"
log2 "Open_IM/src/common/log"
"github.com/gin-gonic/gin"
sts "github.com/tencentyun/qcloud-cos-sts-sdk/go"
"net/http"
"time"
)
type paramsTencentCloudStorageCredential struct {
Token string `json:"token"`
OperationID string `json:"operationID"`
}
func TencentCloudStorageCredential(c *gin.Context) {
params := paramsTencentCloudStorageCredential{}
if err := c.BindJSON(&params); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"errCode": 400, "errMsg": "Parameter parsing error，please check the parameters and request service again"})
return
}
log2.Info(params.Token, params.OperationID, "api TencentUpLoadCredential call start...")
cli := sts.NewClient(
config.Config.Credential.Tencent.SecretID,
config.Config.Credential.Tencent.SecretKey,
nil,
)
log2.Info(c.Request.Header.Get("token"), c.PostForm("optionID"), "api TencentUpLoadCredential sts.NewClient cli = %v", cli)
opt := &sts.CredentialOptions{
DurationSeconds: int64(time.Hour.Seconds()),
Region: config.Config.Credential.Tencent.Region,
Policy: &sts.CredentialPolicy{
Statement: []sts.CredentialPolicyStatement{
{
Action: []string{
"name/cos:PostObject",
"name/cos:PutObject",
},
Effect: "allow",
Resource: []string{
"qcs::cos:" + config.Config.Credential.Tencent.Region + ":uid/" + config.Config.Credential.Tencent.AppID + ":" + config.Config.Credential.Tencent.Bucket + "/*",
},
},
},
},
}
log2.Info(c.Request.Header.Get("token"), c.PostForm("optionID"), "api TencentUpLoadCredential sts.CredentialOptions opt = %v", opt)
res, err := cli.GetCredential(opt)
if err != nil {
log2.Error(c.Request.Header.Get("token"), c.PostForm("optionID"), "api TencentUpLoadCredential cli.GetCredential err = %s", err.Error())
c.JSON(http.StatusOK, gin.H{
"errCode": config.ErrTencentCredential.ErrCode,
"errMsg": err.Error(),
"data": res,
})
return
}
log2.Info(c.Request.Header.Get("token"), c.PostForm("optionID"), "api TencentUpLoadCredential cli.GetCredential success res = %v, res.Credentials = %v", res, res.Credentials)
c.JSON(http.StatusOK, gin.H{
"errCode": 0,
"errMsg": "",
"data": res,
})
}

150
src/common/config/config.go
Unescape View File

@ -0,0 +1,150 @@
package config
import (
"gopkg.in/yaml.v3"
"io/ioutil"
)
var Config config
type config struct {
ServerIP string `yaml:"serverip"`
Api struct {
GinPort []int `yaml:"openImApiPort"`
}
Credential struct {
Tencent struct {
AppID string `yaml:"appID"`
Region string `yaml:"region"`
Bucket string `yaml:"bucket"`
SecretID string `yaml:"secretID"`
SecretKey string `yaml:"secretKey"`
}
}
Mysql struct {
DBAddress []string `yaml:"dbAddress"`
DBUserName string `yaml:"dbUserName"`
DBPassword string `yaml:"dbPassword"`
DBDatabaseName string `yaml:"dbDatabaseName"`
DBTableName string `yaml:"DBTableName"`
DBMsgTableNum int `yaml:"dbMsgTableNum"`
DBMaxOpenConns int `yaml:"dbMaxOpenConns"`
DBMaxIdleConns int `yaml:"dbMaxIdleConns"`
DBMaxLifeTime int `yaml:"dbMaxLifeTime"`
}
Mongo struct {
DBAddress []string `yaml:"dbAddress"`
DBDirect bool `yaml:"dbDirect"`
DBTimeout int `yaml:"dbTimeout"`
DBDatabase []string `yaml:"dbDatabase"`
DBSource string `yaml:"dbSource"`
DBUserName string `yaml:"dbUserName"`
DBPassword string `yaml:"dbPassword"`
DBMaxPoolSize int `yaml:"dbMaxPoolSize"`
DBRetainChatRecords int `yaml:"dbRetainChatRecords"`
}
Redis struct {
DBAddress []string `yaml:"dbAddress"`
DBMaxIdle int `yaml:"dbMaxIdle"`
DBMaxActive int `yaml:"dbMaxActive"`
DBIdleTimeout int `yaml:"dbIdleTimeout"`
DBPassWord string `yaml:"dbPassWord"`
}
RpcPort struct {
OpenImUserPort []int `yaml:"openImUserPort"`
openImFriendPort []int `yaml:"openImFriendPort"`
RpcMessagePort []int `yaml:"rpcMessagePort"`
RpcPushMessagePort []int `yaml:"rpcPushMessagePort"`
OpenImGroupPort []int `yaml:"openImGroupPort"`
RpcModifyUserInfoPort []int `yaml:"rpcModifyUserInfoPort"`
RpcGetTokenPort []int `yaml:"rpcGetTokenPort"`
}
RpcRegisterName struct {
OpenImUserName string `yaml:"openImUserName"`
OpenImFriendName string `yaml:"openImFriendName"`
OpenImOfflineMessageName string `yaml:"openImOfflineMessageName"`
OpenImPushName string `yaml:"openImPushName"`
OpenImOnlineMessageRelayName string `yaml:"openImOnlineMessageRelayName"`
OpenImGroupName string `yaml:"openImGroupName"`
RpcGetTokenName string `yaml:"rpcGetTokenName"`
}
Etcd struct {
EtcdSchema string `yaml:"etcdSchema"`
EtcdAddr []string `yaml:"etcdAddr"`
}
Log struct {
StorageLocation string `yaml:"storageLocation"`
ElasticSearchSwitch bool `yaml:"elasticSearchSwitch"`
ElasticSearchAddr []string `yaml:"elasticSearchAddr"`
ElasticSearchUser string `yaml:"elasticSearchUser"`
ElasticSearchPassword string `yaml:"elasticSearchPassword"`
}
ModuleName struct {
LongConnSvrName string `yaml:"longConnSvrName"`
MsgTransferName string `yaml:"msgTransferName"`
PushName string `yaml:"pushName"`
}
LongConnSvr struct {
WebsocketPort []int `yaml:"websocketPort"`
WebsocketMaxConnNum int `yaml:"websocketMaxConnNum"`
WebsocketMaxMsgLen int `yaml:"websocketMaxMsgLen"`
WebsocketTimeOut int `yaml:"websocketTimeOut"`
}
Push struct {
Tpns struct {
Ios struct {
AccessID string `yaml:"accessID"`
SecretKey string `yaml:"secretKey"`
}
Android struct {
AccessID string `yaml:"accessID"`
SecretKey string `yaml:"secretKey"`
}
}
}
Kafka struct {
Ws2mschat struct {
Addr []string `yaml:"addr"`
Topic string `yaml:"topic"`
}
Ms2pschat struct {
Addr []string `yaml:"addr"`
Topic string `yaml:"topic"`
}
ConsumerGroupID struct {
MsgToMongo string `yaml:"msgToMongo"`
MsgToMySql string `yaml:"msgToMySql"`
MsgToPush string `yaml:"msgToPush"`
}
}
Secret string
MultiLoginPolicy struct {
OnlyOneTerminalAccess bool `yaml:"onlyOneTerminalAccess"`
MobileAndPCTerminalAccessButOtherTerminalKickEachOther bool `yaml:"mobileAndPCTerminalAccessButOtherTerminalKickEachOther"`
AllTerminalAccess bool `yaml:"allTerminalAccess"`
}
TokenPolicy struct {
AccessSecret string `yaml:"accessSecret"`
AccessExpire int64 `yaml:"accessExpire"`
}
}
func init() {
bytes, err := ioutil.ReadFile("../config/config.yaml")
if err != nil {
panic(err)
return
}
if err = yaml.Unmarshal(bytes, &Config); err != nil {
panic(err)
return
}
}

38
src/common/config/error.go
Unescape View File

@ -0,0 +1,38 @@
package config
// key = errCode, string = errMsg
type ErrInfo struct {
ErrCode int32
ErrMsg string
}
var (
OK = ErrInfo{0, ""}
ErrMysql = ErrInfo{100, ""}
ErrMongo = ErrInfo{110, ""}
ErrRedis = ErrInfo{120, ""}
ErrParseToken = ErrInfo{200, "Parse token failed"}
ErrCreateToken = ErrInfo{201, "Create token failed"}
ErrAppServerKey = ErrInfo{300, "key error"}
ErrTencentCredential = ErrInfo{400, ""}
ErrorUserRegister = ErrInfo{600, "User registration failed"}
ErrAccountExists = ErrInfo{601, "The account is already registered and cannot be registered again"}
ErrUserPassword = ErrInfo{602, "User password error"}
ErrTokenIncorrect = ErrInfo{603, "Invalid token"}
ErrTokenExpired = ErrInfo{604, "Expired token"}
ErrRefreshToken = ErrInfo{605, "Failed to refresh token"}
ErrAddFriend = ErrInfo{606, "Failed to add friends"}
ErrAgreeToAddFriend = ErrInfo{607, "Failed to agree to add friend"}
ErrAddFriendToBlack = ErrInfo{608, "Failed to add friends to the blacklist"}
ErrGetBlackList = ErrInfo{609, "Failed to get blacklist"}
ErrDeleteFriend = ErrInfo{610, "Failed to delete friend"}
ErrGetFriendApplyList = ErrInfo{611, "Failed to get friend application list"}
ErrGetFriendList = ErrInfo{612, "Failed to get friend list"}
ErrRemoveBlackList = ErrInfo{613, "Failed to remove blacklist"}
ErrSearchUserInfo = ErrInfo{614, "Failed to find user information"}
ErrDelAppleDeviceToken = ErrInfo{615, ""}
ErrModifyUserInfo = ErrInfo{616, "update user some attribute failed"}
ErrSetFriendComment = ErrInfo{617, "set friend comment failed"}
)

72
src/common/db/model.go
Unescape View File

@ -0,0 +1,72 @@
package db
import (
"gopkg.in/mgo.v2"
)
var DB DataBases
type DataBases struct {
MgoDB mongoDB
RedisDB redisDB
MysqlDB mysqlDB
}
func key(dbAddress, dbName string) string {
return dbAddress + "_" + dbName
}
//type Config struct {
// Mongo struct {
// DBAddress []string `yaml:"dbAddress"`
// DBDirect bool `yaml:"dbDirect"`
// DBTimeout int `yaml:"dbTimeout"`
// DBDatabase []string `yaml:"dbDatabase"`
// DBSource string `yaml:"dbSource"`
// DBUserName string `yaml:"dbUserName"`
// DBPassword string `yaml:"dbPassword"`
// DBMaxPoolSize int `yaml:"dbMaxPoolSize"`
// }
// Mysql struct {
// DBAddress []string `yaml:"dbAddress"`
// DBPort int `yaml:"dbPort"`
// DBUserName string `yaml:"dbUserName"`
// DBPassword string `yaml:"dbPassword"`
// DBDatabaseName string `yaml:"dbChatName"` // 默认使用DBAddress[0]
// DBTableName string `yaml:"dbMsgName"`
// DBMsgTableNum int `yaml:"dbMsgTableNum"`
// DBCharset string `yaml:"dbCharset"`
// DBMaxOpenConns int `yaml:"dbMaxOpenConns"`
// DBMaxIdleConns int `yaml:"dbMaxIdleConns"`
// DBMaxLifeTime int `yaml:"dbMaxLifeTime"`
// }
// Redis struct {
// DBAddress string `yaml:"dbAddress"`
// DBPort int `yaml:"dbPort"`
// DBMaxIdle int `yaml:"dbMaxIdle"`
// DBMaxActive int `yaml:"dbMaxActive"`
// DBIdleTimeout int `yaml:"dbIdleTimeout"`
// }
//}
//func init() {
// bytes, err := ioutil.ReadFile("config/db.yaml")
// if err != nil {
// log.Error("", "", "read db.yaml config fail! err = %s", err.Error())
// return
// }
//
// if err = yaml.Unmarshal(bytes, &DB.Config); err != nil {
// log.Error("", "", "unmarshal db.yaml config fail! err = %s", err.Error())
// return
// }
//
// DB.RedisDB.newPool(DB.Config)
// //DB.MysqlDB.sqlxDB(DB.Config.Mysql.DBName[0], DB.Config)
//}
func init() {
DB.RedisDB.newPool()
}
func (d *DataBases) session(dbName string) *mgo.Session {
return d.MgoDB.mgoSession(dbName)
}

51
src/common/db/mongo.go
Unescape View File

@ -0,0 +1,51 @@
package db
import (
"Open_IM/src/common/config"
"errors"
"fmt"
"gopkg.in/mgo.v2"
"sync"
"time"
)
type mongoDB struct {
sync.RWMutex
dbMap map[string]*mgo.Session
}
func (m *mongoDB) mgoSession(dbName string) *mgo.Session {
m.Lock()
defer m.Unlock()
if _, ok := m.dbMap[dbName]; !ok {
if err := m.newMgoSession(dbName); err != nil {
panic(err)
return nil
}
}
return m.dbMap[dbName]
}
func (m *mongoDB) newMgoSession(dbName string) error {
dailInfo := &mgo.DialInfo{
Addrs: config.Config.Mongo.DBAddress,
Direct: config.Config.Mongo.DBDirect,
Timeout: time.Second * time.Duration(config.Config.Mongo.DBTimeout),
Database: dbName,
Source: config.Config.Mongo.DBSource,
Username: config.Config.Mongo.DBUserName,
Password: config.Config.Mongo.DBPassword,
PoolLimit: config.Config.Mongo.DBMaxPoolSize,
}
session, err := mgo.DialWithInfo(dailInfo)
if err != nil {
return errors.New(fmt.Sprintf("mongo DialWithInfo fail, err= %s", err.Error()))
}
if m.dbMap == nil {
m.dbMap = make(map[string]*mgo.Session)
}
m.dbMap[dbName] = session
return nil
}

189
src/common/db/mongoModel.go
Unescape View File

@ -0,0 +1,189 @@
package db
import (
"Open_IM/src/common/config"
"Open_IM/src/common/constant"
pbMsg "Open_IM/src/proto/chat"
"errors"
"github.com/golang/protobuf/proto"
"gopkg.in/mgo.v2"
"gopkg.in/mgo.v2/bson"
"time"
)
type UserChat struct {
UID string
Msg [][]byte
}
func (d *DataBases) GetUserChat(uid string, seqBegin, seqEnd int64) (SingleMsg []*pbMsg.MsgFormat, GroupMsg []*pbMsg.MsgFormat, MaxSeq int64, MinSeq int64, err error) {
session := d.session(config.Config.Mongo.DBDatabase[0]).Clone()
if session == nil {
return nil, nil, MaxSeq, MinSeq, errors.New("session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase[0]).C("chat")
sChat := UserChat{}
if err = c.Find(bson.M{"uid": uid}).One(&sChat); err != nil {
return nil, nil, MaxSeq, MinSeq, err
}
pChat := pbMsg.MsgSvrToPushSvrChatMsg{}
for i := 0; i < len(sChat.Msg); i++ {
//每次产生新的指针
temp := new(pbMsg.MsgFormat)
if err = proto.Unmarshal(sChat.Msg[i], &pChat); err != nil {
return nil, nil, MaxSeq, MinSeq, err
}
if pChat.RecvSeq >= seqBegin && pChat.RecvSeq <= seqEnd {
temp.SendID = pChat.SendID
temp.RecvID = pChat.RecvID
temp.MsgFrom = pChat.MsgFrom
temp.Seq = pChat.RecvSeq
temp.ServerMsgID = pChat.MsgID
temp.SendTime = pChat.SendTime
temp.Content = pChat.Content
temp.ContentType = pChat.ContentType
temp.SenderPlatformID = pChat.PlatformID
if pChat.RecvSeq > MaxSeq {
MaxSeq = pChat.RecvSeq
}
if i == 0 {
MinSeq = pChat.RecvSeq
}
if pChat.RecvSeq < MinSeq {
MinSeq = pChat.RecvSeq
}
//单聊消息
if pChat.SessionType == constant.SingleChatType {
SingleMsg = append(SingleMsg, temp)
} else {
GroupMsg = append(GroupMsg, temp)
}
}
}
//d.DelUserChat(&sChat)
return SingleMsg, GroupMsg, MaxSeq, MinSeq, nil
}
func (d *DataBases) SaveUserChat(uid string, m proto.Message) error {
session := d.session(config.Config.Mongo.DBDatabase[0]).Clone()
if session == nil {
return errors.New("session == nil")
}
defer session.Close()
session.SetMode(mgo.Monotonic, true)
c := session.DB(config.Config.Mongo.DBDatabase[0]).C("chat")
n, err := c.Find(bson.M{"uid": uid}).Count()
if err != nil {
return err
}
if n == 0 {
sChat := UserChat{}
sChat.UID = uid
bMsg, _ := proto.Marshal(m)
sChat.Msg = append(sChat.Msg, bMsg)
err = c.Insert(&sChat)
if err != nil {
return err
}
} else {
bMsg, err := proto.Marshal(m)
err = c.Update(bson.M{"uid": uid}, bson.M{"$addToSet": bson.M{"msg": bMsg}})
if err != nil {
return err
}
}
return nil
}
func (d *DataBases) DelUserChat(uc *UserChat) {
delMaxIndex := 0
pbData := pbMsg.WSToMsgSvrChatMsg{}
for i := 0; i < len(uc.Msg); i++ {
if err := proto.Unmarshal(uc.Msg[i], &pbData); err != nil {
delMaxIndex = i
} else {
if time.Now().Unix()-pbData.SendTime > 7*24*3600 {
delMaxIndex = i
} else {
break
}
}
}
if delMaxIndex > 0 {
uc.Msg = uc.Msg[delMaxIndex:]
session := d.session(config.Config.Mongo.DBDatabase[0]).Clone()
if session == nil {
return
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase[0]).C("chat")
if err := c.Update(bson.M{"uid": uc.UID}, bson.M{"msg": uc.Msg}); err != nil {
return
}
}
}
func (d *DataBases) DelHistoryChat(days int64, ids []string) error {
session := d.session(config.Config.Mongo.DBDatabase[0]).Clone()
if session == nil {
return errors.New("mgo session == nil")
}
defer session.Close()
c := session.DB(config.Config.Mongo.DBDatabase[0]).C("chat")
for i := 0; i < len(ids); i++ {
d.delHistoryUserChat(c, days, ids[i])
//time.Sleep(1 * time.Millisecond)
}
return nil
}
func (d *DataBases) delHistoryUserChat(c *mgo.Collection, days int64, id string) error {
sChat := UserChat{}
if err := c.Find(bson.M{"uid": id}).One(&sChat); err != nil {
return err
}
delMaxIndex := 0
pbData := pbMsg.WSToMsgSvrChatMsg{}
for i := 0; i < len(sChat.Msg); i++ {
if err := proto.Unmarshal(sChat.Msg[i], &pbData); err != nil {
delMaxIndex = i
} else {
if time.Now().Unix()-pbData.SendTime > int64(days)*24*3600 {
delMaxIndex = i
} else {
break
}
}
}
if delMaxIndex > 0 {
if delMaxIndex < len(sChat.Msg) {
sChat.Msg = sChat.Msg[delMaxIndex:]
} else {
sChat.Msg = sChat.Msg[0:0]
}
if err := c.Update(bson.M{"uid": sChat.UID}, bson.M{"msg": sChat.Msg}); err != nil {
return err
}
}
return nil
}

53
src/common/db/mysql.go
Unescape View File

@ -0,0 +1,53 @@
package db
import (
"Open_IM/src/common/config"
"fmt"
"github.com/jinzhu/gorm"
_ "github.com/jinzhu/gorm/dialects/mysql"
"sync"
"time"
)
type mysqlDB struct {
sync.RWMutex
dbMap map[string]*gorm.DB
}
func (m *mysqlDB) DefaultGormDB() (*gorm.DB, error) {
return m.GormDB(config.Config.Mysql.DBAddress[0], config.Config.Mysql.DBDatabaseName)
}
func (m *mysqlDB) GormDB(dbAddress, dbName string) (*gorm.DB, error) {
m.Lock()
defer m.Unlock()
k := key(dbAddress, dbName)
if _, ok := m.dbMap[k]; !ok {
if err := m.open(dbAddress, dbName); err != nil {
return nil, err
}
}
return m.dbMap[k], nil
}
func (m *mysqlDB) open(dbAddress, dbName string) error {
dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=true&loc=Local",
config.Config.Mysql.DBUserName, config.Config.Mysql.DBPassword, dbAddress, dbName)
db, err := gorm.Open("mysql", dsn)
if err != nil {
return err
}
db.SingularTable(true)
db.DB().SetMaxOpenConns(config.Config.Mysql.DBMaxOpenConns)
db.DB().SetMaxIdleConns(config.Config.Mysql.DBMaxIdleConns)
db.DB().SetConnMaxLifetime(time.Duration(config.Config.Mysql.DBMaxLifeTime) * time.Second)
if m.dbMap == nil {
m.dbMap = make(map[string]*gorm.DB)
}
k := key(dbAddress, dbName)
m.dbMap[k] = db
return nil
}

52
src/common/db/redis.go
Unescape View File

@ -0,0 +1,52 @@
package db
import (
"Open_IM/src/common/config"
log2 "Open_IM/src/common/log"
"github.com/garyburd/redigo/redis"
"time"
)
type redisDB struct {
pool *redis.Pool
}
func (r *redisDB) newPool() {
r.pool = &redis.Pool{
MaxIdle: config.Config.Redis.DBMaxIdle,
MaxActive: config.Config.Redis.DBMaxActive,
IdleTimeout: time.Duration(config.Config.Redis.DBIdleTimeout) * time.Second,
Dial: func() (redis.Conn, error) {
return redis.Dial(
"tcp",
config.Config.Redis.DBAddress[0],
redis.DialReadTimeout(time.Duration(1000)*time.Millisecond),
redis.DialWriteTimeout(time.Duration(1000)*time.Millisecond),
redis.DialConnectTimeout(time.Duration(1000)*time.Millisecond),
redis.DialDatabase(0),
redis.DialPassword(config.Config.Redis.DBPassWord),
)
},
}
}
func (r *redisDB) Exec(cmd string, key interface{}, args ...interface{}) (interface{}, error) {
con := r.pool.Get()
if err := con.Err(); err != nil {
log2.Error("", "", "redis cmd = %v, err = %v", cmd, err)
return nil, err
}
defer con.Close()
params := make([]interface{}, 0)
params = append(params, key)
if len(args) > 0 {
for _, v := range args {
params = append(params, v)
}
}
return con.Do(cmd, params...)
}

71
src/common/db/redisModel.go
Unescape View File

@ -0,0 +1,71 @@
package db
import (
"github.com/garyburd/redigo/redis"
)
const (
userIncrSeq = "REDIS_USER_INCR_SEQ:" // user incr seq
appleDeviceToken = "DEVICE_TOKEN"
lastGetSeq = "LAST_GET_SEQ"
)
//执行用户消息的seq自增操作
func (d *DataBases) IncrUserSeq(uid string) (int64, error) {
key := userIncrSeq + uid
return redis.Int64(d.RedisDB.Exec("INCR", key))
}
//获取最新的seq
func (d *DataBases) GetUserSeq(uid string) (int64, error) {
key := userIncrSeq + uid
return redis.Int64(d.RedisDB.Exec("GET", key))
}
//存储苹果的设备token到redis
func (d *DataBases) SetAppleDeviceToken(accountAddress, value string) (err error) {
key := appleDeviceToken + accountAddress
_, err = d.RedisDB.Exec("SET", key, value)
return err
}
//删除苹果设备token
func (d *DataBases) DelAppleDeviceToken(accountAddress string) (err error) {
key := appleDeviceToken + accountAddress
_, err = d.RedisDB.Exec("DEL", key)
return err
}
//记录用户上一次主动拉取Seq的值
func (d *DataBases) SetLastGetSeq(uid string) (err error) {
key := lastGetSeq + uid
_, err = d.RedisDB.Exec("SET", key)
return err
}
//获取用户上一次主动拉取Seq的值
func (d *DataBases) GetLastGetSeq(uid string) (int64, error) {
key := userIncrSeq + uid
return redis.Int64(d.RedisDB.Exec("GET", key))
}
//Store userid and platform class to redis
func (d *DataBases) SetUserIDAndPlatform(userID, platformClass, value string, ttl int64) error {
key := userID + platformClass
_, err := d.RedisDB.Exec("SET", key, value, "EX", ttl)
return err
}
//Check exists userid and platform class from redis
func (d *DataBases) ExistsUserIDAndPlatform(userID, platformClass string) (interface{}, error) {
key := userID + platformClass
exists, err := d.RedisDB.Exec("EXISTS", key)
return exists, err
}
//Get platform class Token
func (d *DataBases) GetPlatformToken(userID, platformClass string) (interface{}, error) {
key := userID + platformClass
token, err := d.RedisDB.Exec("GET", key)
return token, err
}
Write Preview
Loading…
Cancel
Save