Merge branch 'main' into pr_branch

pull/25/head
Away 4 years ago committed by GitHub
commit 2722baf577
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

4
.gitignore vendored

@ -0,0 +1,4 @@
components
logs

@ -60,7 +60,7 @@ serverip:
api:
openImApiPort: [ 10000 ]
sdk:
sdkWsPort: [ 30000 ]
openImSdkWsPort: [ 30000 ]
credential:
tencent:
@ -88,12 +88,13 @@ rpcregistername:
openImPushName: Push
openImOnlineMessageRelayName: OnlineMessageRelay
openImGroupName: Group
rpcGetTokenName: Auth
openImAuthName: Auth
log:
storageLocation: ../logs/
rotationTime: 12
remainRotationCount: 10
rotationTime: 24
remainRotationCount: 5
remainLogLevel: 6
elasticSearchSwitch: false
elasticSearchAddr: [ 127.0.0.1:9201 ]
elasticSearchUser: ""
@ -105,7 +106,7 @@ modulename:
pushName: push
longconnsvr:
websocketPort: [ 17778 ]
openImWsPort: [ 17778 ]
websocketMaxConnNum: 10000
websocketMaxMsgLen: 4096
websocketTimeOut: 10

@ -28,7 +28,7 @@ RUN ln -fs /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && dpkg-reconfigure
#set directory to map logs,config file,script file.
VOLUME ["/Open-IM-Server/logs","/Open-IM-Server/config","/Open-IM-Server/script"]
VOLUME ["/Open-IM-Server/logs","/Open-IM-Server/config","/Open-IM-Server/script","/Open-IM-Server/db/sdk"]
#Copy scripts files and binary files to the blank image
COPY --from=build /Open-IM-Server/script /Open-IM-Server/script
@ -36,4 +36,4 @@ COPY --from=build /Open-IM-Server/bin /Open-IM-Server/bin
WORKDIR /Open-IM-Server/script
CMD ["./docker_start_all.sh"]
CMD ["./docker_start_all.sh"]

@ -15,7 +15,7 @@ services:
restart: always
mongodb:
image: mongo
image: mongo:4.0
ports:
- 27017:27017
container_name: mongo
@ -84,25 +84,28 @@ services:
#fixme-----build from dockerfile---------
# open-im-server:
# image: open_im_server
# ports:
# - 10000:10000
# - 17778:17778
# container_name: open-im-server
# volumes:
# - ./logs:/Open-IM-Server/logs
# - ./config/config.yaml:/Open-IM-Server/config/config.yaml
# restart: always
# build:
# context: .
# dockerfile: deploy.Dockerfile
# depends_on:
# - mysql
# - mongodb
# - redis
# - kafka
# - etcd
# open-im-server:
# image: open_im_server
# container_name: open-im-server
# volumes:
# - ./logs:/Open-IM-Server/logs
# - ./config/config.yaml:/Open-IM-Server/config/config.yaml
# restart: always
# build:
# context: .
# dockerfile: deploy.Dockerfile
# depends_on:
# - mysql
# - mongodb
# - redis
# - kafka
# - etcd
# network_mode: "host"
# logging:
# driver: json-file
# options:
# max-size: "1g"
# max-file: "2"
#fixme----build from docker hub------
open-im-server:

@ -32,6 +32,7 @@ require (
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
github.com/rifflock/lfshook v0.0.0-20180920164130-b9218ef580f5
github.com/sirupsen/logrus v1.6.0
github.com/stretchr/testify v1.7.0
github.com/tencentyun/qcloud-cos-sts-sdk v0.0.0-20210325043845-84a0811633ca
github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 // indirect
go.etcd.io/etcd v0.0.0-20200402134248-51bdeb39e698

@ -235,6 +235,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/tencentyun/qcloud-cos-sts-sdk v0.0.0-20210325043845-84a0811633ca h1:G/aIr3WiUesWHL2YGYgEqjM5tCAJ43Ml+0C18wDkWWs=
github.com/tencentyun/qcloud-cos-sts-sdk v0.0.0-20210325043845-84a0811633ca/go.mod h1:b18KQa4IxHbxeseW1GcZox53d7J0z39VNONTxvvlkXw=
github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=

@ -4,9 +4,7 @@ source ./style_info.cfg
source ./path_info.cfg
source ./function.sh
service_port_name=(
#api port name
openImApiPort
#rpc port name
openImUserPort
openImFriendPort
openImOfflineMessagePort
@ -14,8 +12,8 @@ service_port_name=(
openImGroupPort
openImAuthPort
openImPushPort
websocketPort
sdkWsPort
openImWsPort
openImSdkWsPort
)
for i in ${service_port_name[*]};do
list=$(cat $config_path | grep -w ${i} | awk -F '[:]' '{print $NF}')

@ -8,6 +8,7 @@ need_to_start_server_shell=(
msg_gateway_start.sh
push_start.sh
msg_transfer_start.sh
sdk_svr_start.sh
)
#fixme The 10 second delay to start the project is for the docker-compose one-click to start openIM when the infrastructure dependencies are not started

@ -4,7 +4,7 @@ source ./style_info.cfg
source ./path_info.cfg
source ./function.sh
list1=$(cat $config_path | grep openImOnlineRelayPort | awk -F '[:]' '{print $NF}')
list2=$(cat $config_path | grep websocketPort | awk -F '[:]' '{print $NF}')
list2=$(cat $config_path | grep openImWsPort | awk -F '[:]' '{print $NF}')
list_to_string $list1
rpc_ports=($ports_array)
list_to_string $list2
@ -26,7 +26,7 @@ fi
sleep 1
cd ${msg_gateway_binary_root}
for ((i = 0; i < ${#ws_ports[@]}; i++)); do
nohup ./${msg_gateway_name} -rpc_port ${rpc_ports[$i]} -ws_port ${ws_ports[$i]} >>../logs/${msg_gateway_name}.log 2>&1 &
nohup ./${msg_gateway_name} -rpc_port ${rpc_ports[$i]} -ws_port ${ws_ports[$i]} >>../logs/openIM.log 2>&1 &
done
#Check launched service process
@ -39,7 +39,7 @@ if [ $check -ge 1 ]; then
ports=$(netstat -netulp | grep -w ${i} | awk '{print $4}' | awk -F '[:]' '{print $NF}')
allPorts=${allPorts}"$ports "
done
echo -e ${SKY_BLUE_PREFIX}"SERVICE START SUCCESS !!!"${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"SERVICE START SUCCESS"${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"SERVICE_NAME: "${COLOR_SUFFIX}${YELLOW_PREFIX}${msg_gateway_name}${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"PID: "${COLOR_SUFFIX}${YELLOW_PREFIX}${allNewPid}${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"LISTENING_PORT: "${COLOR_SUFFIX}${YELLOW_PREFIX}${allPorts}${COLOR_SUFFIX}

@ -16,7 +16,7 @@ fi
#Waiting port recycling
sleep 1
cd ${msg_transfer_binary_root}
nohup ./${msg_transfer_name} >>../logs/${msg_transfer_name}.log 2>&1 &
nohup ./${msg_transfer_name} >>../logs/openIM.log 2>&1 &
#Check launched service process
check=`ps aux | grep -w ./${msg_transfer_name} | grep -v grep| wc -l`
if [ $check -eq 1 ]
@ -29,7 +29,7 @@ for i in $ports ;
do
allPorts=${allPorts}"$i "
done
echo -e ${SKY_BLUE_PREFIX}"SERVICE START SUCCESS !!!"${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"SERVICE START SUCCESS "${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"SERVICE_NAME: "${COLOR_SUFFIX}${YELLOW_PREFIX}${msg_transfer_name}${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"PID: "${COLOR_SUFFIX}${YELLOW_PREFIX}${newPid}${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"LISTENING_PORT: "${COLOR_SUFFIX}${YELLOW_PREFIX}${allPorts}${COLOR_SUFFIX}

@ -22,7 +22,7 @@ sleep 1
cd ${push_binary_root}
for ((i = 0; i < ${#rpc_ports[@]}; i++)); do
nohup ./${push_name} -port ${rpc_ports[$i]} >>../logs/${push_name}.log 2>&1 &
nohup ./${push_name} -port ${rpc_ports[$i]} >>../logs/openIM.log 2>&1 &
done
sleep 3
@ -36,7 +36,7 @@ if [ $check -eq 1 ]; then
for i in $ports; do
allPorts=${allPorts}"$i "
done
echo -e ${SKY_BLUE_PREFIX}"SERVICE START SUCCESS !!!"${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"SERVICE START SUCCESS "${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"SERVICE_NAME: "${COLOR_SUFFIX}${YELLOW_PREFIX}${push_name}${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"PID: "${COLOR_SUFFIX}${YELLOW_PREFIX}${newPid}${COLOR_SUFFIX}
echo -e ${SKY_BLUE_PREFIX}"LISTENING_PORT: "${COLOR_SUFFIX}${YELLOW_PREFIX}${allPorts}${COLOR_SUFFIX}

@ -4,8 +4,8 @@ source ./style_info.cfg
source ./path_info.cfg
source ./function.sh
list1=$(cat $config_path | grep openImApiPort | awk -F '[:]' '{print $NF}')
list2=$(cat $config_path | grep websocketPort | awk -F '[:]' '{print $NF}')
list3=$(cat $config_path | grep sdkWsPort | awk -F '[:]' '{print $NF}')
list2=$(cat $config_path | grep openImWsPort | awk -F '[:]' '{print $NF}')
list3=$(cat $config_path | grep openImSdkWsPort | awk -F '[:]' '{print $NF}')
list_to_string $list1
api_ports=($ports_array)
list_to_string $list2
@ -26,7 +26,7 @@ fi
#Waiting port recycling
sleep 1
cd ${sdk_server_binary_root}
nohup ./${sdk_server_name} -openIM_api_port ${api_ports[0]} -openIM_ws_port ${ws_ports[0]} -sdk_ws_port ${sdk_ws_ports[0]} >>../logs/${sdk_server_name}.log 2>&1 &
nohup ./${sdk_server_name} -openIM_api_port ${api_ports[0]} -openIM_ws_port ${ws_ports[0]} -sdk_ws_port ${sdk_ws_ports[0]} >>../logs/openIM.log 2>&1 &
#Check launched service process
sleep 3

@ -8,7 +8,7 @@ need_to_start_server_shell=(
msg_gateway_start.sh
push_start.sh
msg_transfer_start.sh
start_sdk_svr.sh
sdk_svr_start.sh
)
for i in ${need_to_start_server_shell[*]}; do

@ -40,7 +40,7 @@ func newUserRegisterReq(params *paramsUserRegister) *pbAuth.UserRegisterReq {
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)
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImAuthName)
client := pbAuth.NewAuthClient(etcdConn)
//defer etcdConn.Close()

@ -27,7 +27,7 @@ func newUserTokenReq(params *paramsUserToken) *pbAuth.UserTokenReq {
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)
etcdConn := getcdv3.GetConn(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImAuthName)
client := pbAuth.NewAuthClient(etcdConn)
//defer etcdConn.Close()

@ -8,6 +8,7 @@ import (
"Open_IM/src/api/manage"
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"
@ -93,7 +94,7 @@ func main() {
managementGroup.POST("/send_msg", manage.ManagementSendMsg)
managementGroup.POST("/get_all_users_uid", manage.GetAllUsersUid)
}
log.NewPrivateLog("api")
ginPort := flag.Int("port", 10000, "get ginServerPort from cmd,default 10000 as port")
flag.Parse()
r.Run(utils.ServerIP + ":" + strconv.Itoa(*ginPort))

@ -1,8 +1,17 @@
package config
import (
"gopkg.in/yaml.v3"
"io/ioutil"
"path/filepath"
"runtime"
"gopkg.in/yaml.v3"
)
var (
_, b, _, _ = runtime.Caller(0)
// Root folder of this project
Root = filepath.Join(filepath.Dir(b), "../../..")
)
var Config config
@ -14,7 +23,7 @@ type config struct {
GinPort []int `yaml:"openImApiPort"`
}
Sdk struct {
WsPort []int `yaml:"sdkWsPort"`
WsPort []int `yaml:"openImSdkWsPort"`
}
Credential struct {
Tencent struct {
@ -71,7 +80,7 @@ type config struct {
OpenImPushName string `yaml:"openImPushName"`
OpenImOnlineMessageRelayName string `yaml:"openImOnlineMessageRelayName"`
OpenImGroupName string `yaml:"openImGroupName"`
RpcGetTokenName string `yaml:"rpcGetTokenName"`
OpenImAuthName string `yaml:"openImAuthName"`
}
Etcd struct {
EtcdSchema string `yaml:"etcdSchema"`
@ -81,6 +90,7 @@ type config struct {
StorageLocation string `yaml:"storageLocation"`
RotationTime int `yaml:"rotationTime"`
RemainRotationCount uint `yaml:"remainRotationCount"`
RemainLogLevel uint `yaml:"remainLogLevel"`
ElasticSearchSwitch bool `yaml:"elasticSearchSwitch"`
ElasticSearchAddr []string `yaml:"elasticSearchAddr"`
ElasticSearchUser string `yaml:"elasticSearchUser"`
@ -92,7 +102,7 @@ type config struct {
PushName string `yaml:"pushName"`
}
LongConnSvr struct {
WebsocketPort []int `yaml:"websocketPort"`
WebsocketPort []int `yaml:"openImWsPort"`
WebsocketMaxConnNum int `yaml:"websocketMaxConnNum"`
WebsocketMaxMsgLen int `yaml:"websocketMaxMsgLen"`
WebsocketTimeOut int `yaml:"websocketTimeOut"`
@ -146,14 +156,14 @@ type config struct {
}
func init() {
bytes, err := ioutil.ReadFile("../config/config.yaml")
// if we cd Open-IM-Server/src/utils and run go test
// it will panic cannot find config/config.yaml
bytes, err := ioutil.ReadFile(Root + "/config/config.yaml")
if err != nil {
panic(err)
return
}
if err = yaml.Unmarshal(bytes, &Config); err != nil {
panic(err)
return
}
}

@ -66,7 +66,7 @@ func (d *DataBases) SetLastGetSeq(uid string) (err error) {
//获取用户上一次主动拉取Seq的值
func (d *DataBases) GetLastGetSeq(uid string) (int64, error) {
key := userIncrSeq + uid
key := lastGetSeq + uid
return redis.Int64(d.Exec("GET", key))
}

@ -2,6 +2,7 @@ package log
import (
"Open_IM/src/common/config"
"bufio"
"fmt"
nested "github.com/antonfisher/nested-logrus-formatter"
rotatelogs "github.com/lestrrat-go/file-rotatelogs"
@ -29,18 +30,25 @@ func NewPrivateLog(moduleName string) {
func loggerInit(moduleName string) *Logger {
var logger = logrus.New()
//All logs will be printed
logger.SetLevel(logrus.TraceLevel)
//Log Style Setting
logger.SetLevel(logrus.Level(config.Config.Log.RemainLogLevel))
//Close std console output
src, err := os.OpenFile(os.DevNull, os.O_APPEND|os.O_WRONLY, os.ModeAppend)
if err != nil {
panic(err)
}
writer := bufio.NewWriter(src)
logger.SetOutput(writer)
//Log Console Print Style Setting
logger.SetFormatter(&nested.Formatter{
TimestampFormat: "2006-01-02 15:04:05.000",
HideKeys: false,
FieldsOrder: []string{"PID"},
FieldsOrder: []string{"PID", "FilePath", "OperationID"},
})
//File name and line number display hook
logger.AddHook(newFileHook())
//Send logs to elasticsearch hook
if config.Config.Log.ElasticSearchSwitch == true {
if config.Config.Log.ElasticSearchSwitch {
logger.AddHook(newEsHook(moduleName))
}
//Log file segmentation hook
@ -60,13 +68,16 @@ func NewLfsHook(rotationTime time.Duration, maxRemainNum uint, moduleName string
}, &nested.Formatter{
TimestampFormat: "2006-01-02 15:04:05.000",
HideKeys: false,
FieldsOrder: []string{"PID"},
FieldsOrder: []string{"PID", "FilePath", "OperationID"},
})
return lfsHook
}
func initRotateLogs(rotationTime time.Duration, maxRemainNum uint, level string, moduleName string) *rotatelogs.RotateLogs {
if moduleName != "" {
moduleName = moduleName + "."
}
writer, err := rotatelogs.New(
config.Config.Log.StorageLocation+moduleName+"/"+level+"."+"%Y-%m-%d_%H-%M-%S",
config.Config.Log.StorageLocation+moduleName+level+"."+"%Y-%m-%d",
rotatelogs.WithRotationTime(rotationTime),
rotatelogs.WithRotationCount(maxRemainNum),
)
@ -77,54 +88,50 @@ func initRotateLogs(rotationTime time.Duration, maxRemainNum uint, level string,
}
}
//Deprecated
func Info(token, OperationID, format string, args ...interface{}) {
if token == "" && OperationID == "" {
logger.WithFields(logrus.Fields{}).Infof(format, args...)
} else {
logger.WithFields(logrus.Fields{
"token": token,
"OperationID": OperationID,
}).Infof(format, args...)
}
logger.WithFields(logrus.Fields{
"PID": logger.Pid,
"OperationID": OperationID,
}).Infof(format, args...)
}
//Deprecated
func Error(token, OperationID, format string, args ...interface{}) {
if token == "" && OperationID == "" {
logger.WithFields(logrus.Fields{}).Errorf(format, args...)
} else {
logger.WithFields(logrus.Fields{
"token": token,
"OperationID": OperationID,
}).Errorf(format, args...)
}
logger.WithFields(logrus.Fields{
"PID": logger.Pid,
"OperationID": OperationID,
}).Errorf(format, args...)
}
//Deprecated
func Debug(token, OperationID, format string, args ...interface{}) {
if token == "" && OperationID == "" {
logger.WithFields(logrus.Fields{}).Debugf(format, args...)
} else {
logger.WithFields(logrus.Fields{
"token": token,
"OperationID": OperationID,
}).Debugf(format, args...)
}
logger.WithFields(logrus.Fields{
"PID": logger.Pid,
"OperationID": OperationID,
}).Debugf(format, args...)
}
//Deprecated
func Warning(token, OperationID, format string, args ...interface{}) {
if token == "" && OperationID == "" {
logger.WithFields(logrus.Fields{}).Warningf(format, args...)
} else {
logger.WithFields(logrus.Fields{
"token": token,
"OperationID": OperationID,
}).Warningf(format, args...)
}
logger.WithFields(logrus.Fields{
"PID": logger.Pid,
"OperationID": OperationID,
}).Warningf(format, args...)
}
//Deprecated
func InfoByArgs(format string, args ...interface{}) {
logger.WithFields(logrus.Fields{}).Infof(format, args)
}
//Deprecated
func ErrorByArgs(format string, args ...interface{}) {
logger.WithFields(logrus.Fields{}).Errorf(format, args...)
}
@ -132,21 +139,28 @@ func ErrorByArgs(format string, args ...interface{}) {
//Print log information in k, v format,
//kv is best to appear in pairs. tipInfo is the log prompt information for printing,
//and kv is the key and value for printing.
//Deprecated
func InfoByKv(tipInfo, OperationID string, args ...interface{}) {
fields := make(logrus.Fields)
argsHandle(OperationID, fields, args)
logger.WithFields(fields).Info(tipInfo)
}
//Deprecated
func ErrorByKv(tipInfo, OperationID string, args ...interface{}) {
fields := make(logrus.Fields)
argsHandle(OperationID, fields, args)
logger.WithFields(fields).Error(tipInfo)
}
//Deprecated
func DebugByKv(tipInfo, OperationID string, args ...interface{}) {
fields := make(logrus.Fields)
argsHandle(OperationID, fields, args)
logger.WithFields(fields).Debug(tipInfo)
}
//Deprecated
func WarnByKv(tipInfo, OperationID string, args ...interface{}) {
fields := make(logrus.Fields)
argsHandle(OperationID, fields, args)
@ -162,6 +176,30 @@ func argsHandle(OperationID string, fields logrus.Fields, args []interface{}) {
fields[fmt.Sprintf("%v", args[i])] = ""
}
}
fields["operationID"] = OperationID
fields["OperationID"] = OperationID
fields["PID"] = logger.Pid
}
func NewInfo(OperationID string, args ...interface{}) {
logger.WithFields(logrus.Fields{
"OperationID": OperationID,
"PID": logger.Pid,
}).Infoln(args)
}
func NewError(OperationID string, args ...interface{}) {
logger.WithFields(logrus.Fields{
"OperationID": OperationID,
"PID": logger.Pid,
}).Errorln(args)
}
func NewDebug(OperationID string, args ...interface{}) {
logger.WithFields(logrus.Fields{
"OperationID": OperationID,
"PID": logger.Pid,
}).Debugln(args)
}
func NewWarn(OperationID string, args ...interface{}) {
logger.WithFields(logrus.Fields{
"OperationID": OperationID,
"PID": logger.Pid,
}).Warnln(args)
}

@ -2,7 +2,8 @@ package rpcAuth
import (
"Open_IM/src/common/config"
log2 "Open_IM/src/common/log"
"Open_IM/src/common/log"
"Open_IM/src/grpc-etcdv3/getcdv3"
pbAuth "Open_IM/src/proto/auth"
"Open_IM/src/utils"
@ -20,24 +21,25 @@ type rpcAuth struct {
}
func NewRpcAuthServer(port int) *rpcAuth {
log.NewPrivateLog("auth")
return &rpcAuth{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.RpcGetTokenName,
rpcRegisterName: config.Config.RpcRegisterName.OpenImAuthName,
etcdSchema: config.Config.Etcd.EtcdSchema,
etcdAddr: config.Config.Etcd.EtcdAddr,
}
}
func (rpc *rpcAuth) Run() {
log2.Info("", "", "rpc get_token init...")
log.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)
log.Error("", "", "listen network failed, err = %s, address = %s", err.Error(), address)
return
}
log2.Info("", "", "listen network success, address = %s", address)
log.Info("", "", "listen network success, address = %s", address)
//grpc server
srv := grpc.NewServer()
@ -48,14 +50,14 @@ func (rpc *rpcAuth) Run() {
pbAuth.RegisterAuthServer(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())
log.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())
log.Info("", "", "rpc get_token fail, err = %s", err.Error())
return
}
log2.Info("", "", "rpc get_token init success")
log.Info("", "", "rpc get_token init success")
}

@ -3,7 +3,7 @@ package rpcChat
import (
"Open_IM/src/common/config"
"Open_IM/src/common/kafka"
log2 "Open_IM/src/common/log"
"Open_IM/src/common/log"
"Open_IM/src/grpc-etcdv3/getcdv3"
pbChat "Open_IM/src/proto/chat"
"Open_IM/src/utils"
@ -22,6 +22,7 @@ type rpcChat struct {
}
func NewRpcChatServer(port int) *rpcChat {
log.NewPrivateLog("msg")
rc := rpcChat{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImOfflineMessageName,
@ -33,15 +34,15 @@ func NewRpcChatServer(port int) *rpcChat {
}
func (rpc *rpcChat) Run() {
log2.Info("", "", "rpc get_token init...")
log.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)
log.Error("", "", "listen network failed, err = %s, address = %s", err.Error(), address)
return
}
log2.Info("", "", "listen network success, address = %s", address)
log.Info("", "", "listen network success, address = %s", address)
//grpc server
srv := grpc.NewServer()
@ -52,14 +53,14 @@ func (rpc *rpcChat) Run() {
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())
log.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())
log.Info("", "", "rpc get_token fail, err = %s", err.Error())
return
}
log2.Info("", "", "rpc get_token init success")
log.Info("", "", "rpc get_token init success")
}

@ -88,84 +88,77 @@ func (rpc *rpcChat) UserSendMsg(_ context.Context, pb *pbChat.UserSendMsgReq) (*
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)
}
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.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)
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)
} 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)
for _, v := range kickContent.UidListInfo {
addUidList = append(addUidList, v.UserId)
}
}
return returnMsg(&replay, pb, 0, "", serverMsgID, pbData.SendTime)
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)
}

@ -110,6 +110,10 @@ func (s *friendServer) ImportFriend(ctx context.Context, req *pbFriend.ImportFri
SessionType: constant.SingleChatType,
OperationID: req.OperationID,
})
} else {
resp.CommonResp.ErrorMsg = "some uid establish failed"
resp.CommonResp.ErrorCode = 408
resp.FailedUidList = append(resp.FailedUidList, v)
}
}
}

@ -24,6 +24,7 @@ type friendServer struct {
}
func NewFriendServer(port int) *friendServer {
log.NewPrivateLog("friend")
return &friendServer{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImFriendName,

@ -28,6 +28,7 @@ type groupServer struct {
}
func NewGroupServer(port int) *groupServer {
log.NewPrivateLog("group")
return &groupServer{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImGroupName,

@ -81,7 +81,7 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
log.Error(claims.UID, req.OperationID, "FindUserByUID failed, err: ", err.Error())
return &pbGroup.InviteUserToGroupResp{ErrorCode: config.ErrParam.ErrCode, ErrorMsg: config.ErrParam.ErrMsg}, nil
}*/
var nicknameList string
for _, v := range req.UidList {
var resultNode pbGroup.Id2Result
resultNode.UId = v
@ -113,7 +113,7 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
if err != nil {
log.Error("", "", "add mongo group member failed, db.DB.AddGroupMember fail [err: %s]", err.Error())
}
nicknameList = nicknameList + toUserInfo.Name + " "
resp.Id2Result = append(resp.Id2Result, &resultNode)
}
resp.ErrorCode = 0
@ -121,12 +121,13 @@ func (s *groupServer) InviteUserToGroup(ctx context.Context, req *pbGroup.Invite
//if claims.UID == config.Config.AppManagerUid
if utils.IsContain(claims.UID, config.Config.Manager.AppManagerUid) {
m, _ := imdb.FindUserByUID(claims.UID)
var iu inviteUserToGroupReq
iu.GroupID = req.GroupID
iu.OperationID = req.OperationID
iu.Reason = req.Reason
iu.UidList = req.UidList
n := content_struct.NotificationContent{1, req.GroupID, iu.ContentToString()}
n := content_struct.NotificationContent{1, nicknameList + " invited into the group chat by " + m.Name, iu.ContentToString()}
logic.SendMsgByWS(&pbChat.WSToMsgSvrChatMsg{
SendID: claims.UID,
RecvID: req.GroupID,

@ -22,6 +22,7 @@ type userServer struct {
}
func NewUserServer(port int) *userServer {
log.NewPrivateLog("user")
return &userServer{
rpcPort: port,
rpcRegisterName: config.Config.RpcRegisterName.OpenImUserName,

@ -45,16 +45,17 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbUser.UpdateUserI
RpcResp, err := client.GetFriendList(context.Background(), newReq)
if err != nil {
log.Error(req.Token, req.OperationID, "err=%s,call get friend list rpc server failed", err)
log.ErrorByKv("get friend list rpc server failed", req.OperationID, "err", err.Error(), "req", req.String())
return &pbUser.CommonResp{}, nil
}
if RpcResp.ErrorCode != 0 {
log.ErrorByKv("get friend list rpc server failed", req.OperationID, "err", err.Error(), "req", req.String())
return &pbUser.CommonResp{}, nil
}
self, err := im_mysql_model.FindUserByUID(ownerUid)
if err != nil {
log.ErrorByKv("get self info failed", req.OperationID, "err", err.Error(), "req", req.String())
return &pbUser.CommonResp{}, nil
}
var name, faceUrl string
if self != nil {

@ -13,23 +13,14 @@ func init() {
ServerIP = config.Config.ServerIP
return
}
//fixme Get the ip of the local network card
netInterfaces, err := net.Interfaces()
// see https://gist.github.com/jniltinho/9787946#gistcomment-3019898
conn, err := net.Dial("udp", "8.8.8.8:80")
if err != nil {
panic(err)
}
for i := 0; i < len(netInterfaces); i++ {
//Exclude useless network cards by judging the net.flag Up flag
if (netInterfaces[i].Flags & net.FlagUp) != 0 {
address, _ := netInterfaces[i].Addrs()
for _, addr := range address {
if ipNet, ok := addr.(*net.IPNet); ok && !ipNet.IP.IsLoopback() {
if ipNet.IP.To4() != nil {
ServerIP = ipNet.IP.String()
return
}
}
}
}
}
defer conn.Close()
localAddr := conn.LocalAddr().(*net.UDPAddr)
ServerIP = localAddr.IP.String()
}

@ -0,0 +1,12 @@
package utils
import (
"net"
"testing"
)
func TestServerIP(t *testing.T) {
if net.ParseIP(ServerIP) == nil {
t.Fail()
}
}

@ -25,21 +25,16 @@ type Claims struct {
func BuildClaims(uid, platform string, ttl int64) Claims {
now := time.Now().Unix()
//if ttl=-1 Permanent token
if ttl == -1 {
return Claims{
UID: uid,
Platform: platform,
StandardClaims: jwt.StandardClaims{
ExpiresAt: -1,
IssuedAt: now,
NotBefore: now,
}}
expiresAt := int64(-1)
if ttl != -1 {
expiresAt = now + ttl
}
return Claims{
UID: uid,
Platform: platform,
StandardClaims: jwt.StandardClaims{
ExpiresAt: now + ttl, //Expiration time
ExpiresAt: expiresAt, //Expiration time
IssuedAt: now, //Issuing time
NotBefore: now, //Begin Effective time
}}
@ -59,7 +54,7 @@ func secret() jwt.Keyfunc {
}
}
func ParseToken(tokensString string) (claims *Claims, err error) {
func getClaimFromToken(tokensString string) (*Claims, error) {
token, err := jwt.ParseWithClaims(tokensString, &Claims{}, secret())
if err != nil {
if ve, ok := err.(*jwt.ValidationError); ok {
@ -75,73 +70,63 @@ func ParseToken(tokensString string) (claims *Claims, err error) {
}
}
if claims, ok := token.Claims.(*Claims); ok && token.Valid {
// 1.check userid and platform class 0 not exists and 1 exists
existsInterface, err := db.DB.ExistsUserIDAndPlatform(claims.UID, Platform2class[claims.Platform])
return claims, nil
}
return nil, err
}
func ParseToken(tokensString string) (claims *Claims, err error) {
claims, err = getClaimFromToken(tokensString)
if err != nil {
return nil, err
}
// 1.check userid and platform class 0 not exists and 1 exists
existsInterface, err := db.DB.ExistsUserIDAndPlatform(claims.UID, Platform2class[claims.Platform])
if err != nil {
return nil, err
}
exists := existsInterface.(int64)
//get config multi login policy
if config.Config.MultiLoginPolicy.OnlyOneTerminalAccess {
//OnlyOneTerminalAccess policy need to check all terminal
//When only one end is allowed to log in, there is a situation that needs to be paid attention to. After PC login,
//mobile login should check two platform times. One of them is less than the redis storage time, which is the invalid token.
platform := "PC"
if Platform2class[claims.Platform] == "PC" {
platform = "Mobile"
}
existsInterface, err = db.DB.ExistsUserIDAndPlatform(claims.UID, platform)
if err != nil {
return nil, err
}
exists := existsInterface.(int64)
//get config multi login policy
if config.Config.MultiLoginPolicy.OnlyOneTerminalAccess {
//OnlyOneTerminalAccess policy need to check all terminal
//When only one end is allowed to log in, there is a situation that needs to be paid attention to. After PC login,
//mobile login should check two platform times. One of them is less than the redis storage time, which is the invalid token.
if Platform2class[claims.Platform] == "PC" {
existsInterface, err = db.DB.ExistsUserIDAndPlatform(claims.UID, "Mobile")
if err != nil {
return nil, err
}
exists = existsInterface.(int64)
if exists == 1 {
res, err := MakeTheTokenInvalid(*claims, "Mobile")
if err != nil {
return nil, err
}
if res {
return nil, TokenInvalid
}
}
} else {
existsInterface, err = db.DB.ExistsUserIDAndPlatform(claims.UID, "PC")
if err != nil {
return nil, err
}
exists = existsInterface.(int64)
if exists == 1 {
res, err := MakeTheTokenInvalid(*claims, "PC")
if err != nil {
return nil, err
}
if res {
return nil, TokenInvalid
}
}
}
if exists == 1 {
res, err := MakeTheTokenInvalid(*claims, Platform2class[claims.Platform])
if err != nil {
return nil, err
}
if res {
return nil, TokenInvalid
}
exists = existsInterface.(int64)
if exists == 1 {
res, err := MakeTheTokenInvalid(*claims, platform)
if err != nil {
return nil, err
}
} else if config.Config.MultiLoginPolicy.MobileAndPCTerminalAccessButOtherTerminalKickEachOther {
if exists == 1 {
res, err := MakeTheTokenInvalid(*claims, Platform2class[claims.Platform])
if err != nil {
return nil, err
}
if res {
return nil, TokenInvalid
}
if res {
return nil, TokenInvalid
}
}
return claims, nil
}
return nil, TokenUnknown
// config.Config.MultiLoginPolicy.MobileAndPCTerminalAccessButOtherTerminalKickEachOther == true
// or PC/Mobile validate success
// final check
if exists == 1 {
res, err := MakeTheTokenInvalid(*claims, Platform2class[claims.Platform])
if err != nil {
return nil, err
}
if res {
return nil, TokenInvalid
}
}
return claims, nil
}
func MakeTheTokenInvalid(currentClaims Claims, platformClass string) (bool, error) {
@ -159,35 +144,16 @@ func MakeTheTokenInvalid(currentClaims Claims, platformClass string) (bool, erro
}
return false, nil
}
func ParseRedisInterfaceToken(redisToken interface{}) (*Claims, error) {
token, err := jwt.ParseWithClaims(string(redisToken.([]uint8)), &Claims{}, secret())
if err != nil {
if ve, ok := err.(*jwt.ValidationError); ok {
if ve.Errors&jwt.ValidationErrorMalformed != 0 {
return nil, TokenMalformed
} else if ve.Errors&jwt.ValidationErrorExpired != 0 {
return nil, TokenExpired
} else if ve.Errors&jwt.ValidationErrorNotValidYet != 0 {
return nil, TokenNotValidYet
} else {
return nil, TokenInvalid
}
}
}
if claims, ok := token.Claims.(*Claims); ok && token.Valid {
return claims, nil
}
return nil, err
return getClaimFromToken(string(redisToken.([]uint8)))
}
//Validation token, false means failure, true means successful verification
func VerifyToken(token, uid string) bool {
claims, err := ParseToken(token)
if err != nil {
return false
} else if claims.UID != uid {
if err != nil || claims.UID != uid {
return false
} else {
return true
}
return true
}

@ -0,0 +1,86 @@
package utils
import (
"Open_IM/src/common/config"
"testing"
"time"
"github.com/stretchr/testify/assert"
)
func Test_BuildClaims(t *testing.T) {
uid := "1"
accountAddr := "accountAddr"
platform := "PC"
ttl := int64(-1)
claim := BuildClaims(uid, accountAddr, platform, ttl)
now := time.Now().Unix()
assert.Equal(t, claim.UID, uid, "uid should equal")
assert.Equal(t, claim.Platform, platform, "platform should equal")
assert.Equal(t, claim.StandardClaims.ExpiresAt, int64(-1), "StandardClaims.ExpiresAt should be equal")
// time difference within 1s
assert.Equal(t, claim.StandardClaims.IssuedAt, now, "StandardClaims.IssuedAt should be equal")
assert.Equal(t, claim.StandardClaims.NotBefore, now, "StandardClaims.NotBefore should be equal")
ttl = int64(60)
now = time.Now().Unix()
claim = BuildClaims(uid, accountAddr, platform, ttl)
// time difference within 1s
assert.Equal(t, claim.StandardClaims.ExpiresAt, int64(60)+now, "StandardClaims.ExpiresAt should be equal")
assert.Equal(t, claim.StandardClaims.IssuedAt, now, "StandardClaims.IssuedAt should be equal")
assert.Equal(t, claim.StandardClaims.NotBefore, now, "StandardClaims.NotBefore should be equal")
}
func Test_CreateToken(t *testing.T) {
uid := "1"
accountAddr := "accountAddr"
platform := int32(1)
now := time.Now().Unix()
tokenString, expiresAt, err := CreateToken(uid, accountAddr, platform)
assert.NotEmpty(t, tokenString)
assert.Equal(t, expiresAt, 604800+now)
assert.Nil(t, err)
}
func Test_VerifyToken(t *testing.T) {
uid := "1"
accountAddr := "accountAddr"
platform := int32(1)
tokenString, _, _ := CreateToken(uid, accountAddr, platform)
result := VerifyToken(tokenString, uid)
assert.True(t, result)
result = VerifyToken(tokenString, "2")
assert.False(t, result)
}
func Test_ParseRedisInterfaceToken(t *testing.T) {
uid := "1"
accountAddr := "accountAddr"
platform := int32(1)
tokenString, _, _ := CreateToken(uid, accountAddr, platform)
claims, err := ParseRedisInterfaceToken([]uint8(tokenString))
assert.Nil(t, err)
assert.Equal(t, claims.UID, uid)
// timeout
config.Config.TokenPolicy.AccessExpire = -80
tokenString, _, _ = CreateToken(uid, accountAddr, platform)
claims, err = ParseRedisInterfaceToken([]uint8(tokenString))
assert.Equal(t, err, TokenExpired)
assert.Nil(t, claims)
}
func Test_ParseToken(t *testing.T) {
uid := "1"
accountAddr := "accountAddr"
platform := int32(1)
tokenString, _, _ := CreateToken(uid, accountAddr, platform)
claims, err := ParseToken(tokenString)
if err == nil {
assert.Equal(t, claims.UID, uid)
}
}
Loading…
Cancel
Save