diff --git a/config/config.yaml b/config/config.yaml index 724e4db0e..7e5912ea2 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -70,12 +70,12 @@ credential: rpcport: - openImUserPort: [ 10100,10101 ] - openImFriendPort: [ 10200,10201 ] + openImUserPort: [ 10100 ] + openImFriendPort: [ 10200 ] openImOfflineMessagePort: [ 10300 ] openImOnlineRelayPort: [ 10400 ] - openImGroupPort: [ 10500 ,10501 ] - openImAuthPort: [ 10600, 10601 ] + openImGroupPort: [ 10500 ] + openImAuthPort: [ 10600 ] openImPushPort: [ 10700 ] diff --git a/go.mod b/go.mod index 3177f4329..ac0c9edf0 100644 --- a/go.mod +++ b/go.mod @@ -36,9 +36,9 @@ require ( github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 // indirect go.etcd.io/etcd v0.0.0-20200402134248-51bdeb39e698 golang.org/x/image v0.0.0-20210220032944-ac19c3e999fb + golang.org/x/net v0.0.0-20201110031124-69a78807bb2b golang.org/x/tools v0.0.0-20210106214847-113979e3529a // indirect google.golang.org/grpc v1.33.2 - google.golang.org/protobuf v1.26.0 gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c sigs.k8s.io/yaml v1.2.0 // indirect diff --git a/script/msg_gateway_start.sh b/script/msg_gateway_start.sh index fdf53e608..2f9c60215 100644 --- a/script/msg_gateway_start.sh +++ b/script/msg_gateway_start.sh @@ -20,7 +20,7 @@ fi check=$(ps aux | grep -w ./${msg_gateway_name} | grep -v grep | wc -l) if [ $check -ge 1 ]; then oldPid=$(ps aux | grep -w ./${msg_gateway_name} | grep -v grep | awk '{print $2}') - kill -9 ${oldPid} + kill -9 ${oldPid} fi #Waiting port recycling sleep 3 @@ -36,8 +36,8 @@ allPorts="" if [ $check -ge 1 ]; then allNewPid=$(ps aux | grep -w ./${msg_gateway_name} | grep -v grep | awk '{print $2}') for i in $allNewPid; do - ports=$(netstat -netulp | grep ${i} | awk '{print $4}' | awk -F '[:]' '{print $NF}') - allPorts=${allPorts}"$ports " + 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_NAME: "${COLOR_SUFFIX}${YELLOW_PREFIX}${msg_gateway_name}${COLOR_SUFFIX} diff --git a/script/msg_transfer_start.sh b/script/msg_transfer_start.sh index 554e46ec1..82dd8a364 100644 --- a/script/msg_transfer_start.sh +++ b/script/msg_transfer_start.sh @@ -3,31 +3,36 @@ source ./style_info.cfg source ./path_info.cfg + + #Check if the service exists #If it is exists,kill this process -check=$(ps aux | grep -w ./${msg_transfer_name} | grep -v grep | wc -l) -if [ $check -eq 1 ]; then - oldPid=$(ps aux | grep -w ./${msg_transfer_name} | grep -v grep | awk '{print $2}') - kill -9 $oldPid +check=`ps aux | grep -w ./${msg_transfer_name} | grep -v grep| wc -l` +if [ $check -eq 1 ] +then +oldPid=`ps aux | grep -w ./${msg_transfer_name} | grep -v grep|awk '{print $2}'` + kill -9 $oldPid 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/${msg_transfer_name}.log 2>&1 & #Check launched service process -check=$(ps aux | grep -w ./${msg_transfer_name} | grep -v grep | wc -l) -if [ $check -eq 1 ]; then - newPid=$(ps aux | grep -w ./${msg_transfer_name} | grep -v grep | awk '{print $2}') - ports=$(netstat -netulp | grep ${newPid} | awk '{print $4}' | awk -F '[:]' '{print $NF}') - allPorts="" +check=`ps aux | grep -w ./${msg_transfer_name} | grep -v grep| wc -l` +if [ $check -eq 1 ] +then +newPid=`ps aux | grep -w ./${msg_transfer_name} | grep -v grep|awk '{print $2}'` +ports=`netstat -netulp | grep -w ${newPid}|awk '{print $4}'|awk -F '[:]' '{print $NF}'` +allPorts="" - 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_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} +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_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} else - echo -e ${YELLOW_PREFIX}${msg_transfer_name}${COLOR_SUFFIX}${RED_PREFIX}"SERVICE START ERROR !!! PLEASE CHECK ERROR LOG"${COLOR_SUFFIX} + echo -e ${YELLOW_PREFIX}${msg_transfer_name}${COLOR_SUFFIX}${RED_PREFIX}"SERVICE START ERROR !!! PLEASE CHECK ERROR LOG"${COLOR_SUFFIX} fi diff --git a/script/push_start.sh b/script/push_start.sh index 61754b206..5dc8546a7 100644 --- a/script/push_start.sh +++ b/script/push_start.sh @@ -4,6 +4,8 @@ source ./style_info.cfg source ./path_info.cfg source ./function.sh + + list1=$(cat $config_path | grep openImPushPort | awk -F '[:]' '{print $NF}') list_to_string $list1 rpc_ports=($ports_array) @@ -23,11 +25,12 @@ for ((i = 0; i < ${#rpc_ports[@]}; i++)); do nohup ./${push_name} -port ${rpc_ports[$i]} >>../logs/${push_name}.log 2>&1 & done +sleep 1 #Check launched service process check=$(ps aux | grep -w ./${push_name} | grep -v grep | wc -l) if [ $check -eq 1 ]; then newPid=$(ps aux | grep -w ./${push_name} | grep -v grep | awk '{print $2}') - ports=$(netstat -netulp | grep ${newPid} | awk '{print $4}' | awk -F '[:]' '{print $NF}') + ports=$(netstat -netulp | grep -w ${newPid} | awk '{print $4}' | awk -F '[:]' '{print $NF}') allPorts="" for i in $ports; do diff --git a/src/common/db/mysql.go b/src/common/db/mysql.go index d790c1c9a..a7a163686 100644 --- a/src/common/db/mysql.go +++ b/src/common/db/mysql.go @@ -2,6 +2,7 @@ package db import ( "Open_IM/src/common/config" + "Open_IM/src/common/log" "fmt" "github.com/jinzhu/gorm" _ "github.com/jinzhu/gorm/dialects/mysql" @@ -15,10 +16,13 @@ type mysqlDB struct { } func initMysqlDB() { + //When there is no open IM database, connect to the mysql built-in database to create openIM database dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=true&loc=Local", config.Config.Mysql.DBUserName, config.Config.Mysql.DBPassword, config.Config.Mysql.DBAddress[0], "mysql") + db, err := gorm.Open("mysql", dsn) if err != nil { + log.Error("", "", dsn) panic(err) } diff --git a/src/common/db/mysql_model/im_mysql_msg_model/chat_log_model.go b/src/common/db/mysql_model/im_mysql_msg_model/chat_log_model.go index ca8c1c0b8..58a02e751 100644 --- a/src/common/db/mysql_model/im_mysql_msg_model/chat_log_model.go +++ b/src/common/db/mysql_model/im_mysql_msg_model/chat_log_model.go @@ -37,7 +37,7 @@ func InsertMessageToChatLog(msgData pbMsg.WSToMsgSvrChatMsg) error { MsgId: msgData.MsgID, SendID: msgData.SendID, RecvID: msgData.RecvID, - SendTime: utils.UnixSecondToTime(msgData.SendTime), + SendTime: utils.UnixNanoSecondToTime(msgData.SendTime), SessionType: msgData.SessionType, ContentType: msgData.ContentType, MsgFrom: msgData.MsgFrom, diff --git a/src/msg_gateway/gate/ws_server.go b/src/msg_gateway/gate/ws_server.go index 78dd56f67..ffc14680d 100644 --- a/src/msg_gateway/gate/ws_server.go +++ b/src/msg_gateway/gate/ws_server.go @@ -59,7 +59,7 @@ func (ws *WServer) readMsg(conn *websocket.Conn) { for { msgType, msg, err := conn.ReadMessage() if err != nil { - log.ErrorByKv("WS ReadMsg error", "", "userIP", conn.RemoteAddr().String(), "userUid", ws.getUserUid(conn), "error", err, "conn", conn) + log.ErrorByKv("WS ReadMsg error", "", "userIP", conn.RemoteAddr().String(), "userUid", ws.getUserUid(conn), "error", err) ws.delUserConn(conn) return } else { diff --git a/src/msg_transfer/logic/history_msg_handler.go b/src/msg_transfer/logic/history_msg_handler.go index eedf274ac..bea5c20b4 100644 --- a/src/msg_transfer/logic/history_msg_handler.go +++ b/src/msg_transfer/logic/history_msg_handler.go @@ -109,6 +109,7 @@ func (mc *HistoryConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, for msg := range claim.Messages() { log.InfoByKv("kafka get info to mongo", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "chat", string(msg.Value)) mc.msgHandle[msg.Topic](msg.Value, string(msg.Key)) + sess.MarkMessage(msg, "") } return nil } diff --git a/src/msg_transfer/logic/persistent_msg_handler.go b/src/msg_transfer/logic/persistent_msg_handler.go index fbbe8b1e3..2c7df4b7c 100644 --- a/src/msg_transfer/logic/persistent_msg_handler.go +++ b/src/msg_transfer/logic/persistent_msg_handler.go @@ -69,6 +69,7 @@ func (pc *PersistentConsumerHandler) ConsumeClaim(sess sarama.ConsumerGroupSessi for msg := range claim.Messages() { log.InfoByKv("kafka get info to mysql", "", "msgTopic", msg.Topic, "msgPartition", msg.Partition, "chat", string(msg.Value)) pc.msgHandle[msg.Topic](msg.Value, string(msg.Key)) + sess.MarkMessage(msg, "") } return nil } diff --git a/src/rpc/chat/chat/pull_message.go b/src/rpc/chat/chat/pull_message.go index 6801852cf..744ae1a36 100644 --- a/src/rpc/chat/chat/pull_message.go +++ b/src/rpc/chat/chat/pull_message.go @@ -2,6 +2,7 @@ package rpcChat import ( "context" + "github.com/garyburd/redigo/redis" commonDB "Open_IM/src/common/db" "Open_IM/src/common/log" @@ -23,8 +24,12 @@ func (rpc *rpcChat) GetNewSeq(_ context.Context, in *pbMsg.GetNewSeqReq) (*pbMsg resp.ErrMsg = "" return resp, err } else { - log.ErrorByKv("getSeq from redis error", in.OperationID, "args", in.String(), "err", err.Error()) - resp.Seq = 0 + if err == redis.ErrNil { + resp.Seq = 0 + } else { + log.ErrorByKv("getSeq from redis error", in.OperationID, "args", in.String(), "err", err.Error()) + resp.Seq = -1 + } resp.ErrCode = 0 resp.ErrMsg = "" return resp, nil diff --git a/src/rpc/group/group/get_group_applicationList.go b/src/rpc/group/group/get_group_applicationList.go index f926b9c51..72cb6e750 100644 --- a/src/rpc/group/group/get_group_applicationList.go +++ b/src/rpc/group/group/get_group_applicationList.go @@ -13,7 +13,7 @@ func (s *groupServer) GetGroupApplicationList(_ context.Context, pb *group.GetGr reply, err := im_mysql_model.GetGroupApplicationList(pb.UID) if err != nil { log.Error("", "", "rpc GetGroupApplicationList call..., im_mysql_model.GetGroupApplicationList fail [uid: %s] [err: %s]", pb.UID, err.Error()) - return nil, err + return &group.GetGroupApplicationListResp{ErrCode: 701, ErrMsg: "GetGroupApplicationList failed"}, nil } log.Info("", "", "rpc GetGroupApplicationList call..., im_mysql_model.GetGroupApplicationList") diff --git a/src/rpc/group/group/group_application_response.go b/src/rpc/group/group/group_application_response.go index a1591ad62..fc0da6306 100644 --- a/src/rpc/group/group/group_application_response.go +++ b/src/rpc/group/group/group_application_response.go @@ -14,7 +14,7 @@ func (s *groupServer) GroupApplicationResponse(_ context.Context, pb *group.Grou reply, err := im_mysql_model.GroupApplicationResponse(pb) if err != nil { log.Error("", "", "rpc GroupApplicationResponse call..., im_mysql_model.GroupApplicationResponse fail [pb: %s] [err: %s]", pb.String(), err.Error()) - return nil, err + return &group.GroupApplicationResponseResp{ErrCode: 702, ErrMsg: "rpc GroupApplicationResponse failed"}, nil } if pb.HandleResult == 1 { diff --git a/src/rpc/user/user/update_user_info.go b/src/rpc/user/user/update_user_info.go index 46db95d43..f4a2c1233 100644 --- a/src/rpc/user/user/update_user_info.go +++ b/src/rpc/user/user/update_user_info.go @@ -41,18 +41,29 @@ func (s *userServer) UpdateUserInfo(ctx context.Context, req *pbUser.UpdateUserI } if RpcResp.ErrorCode != 0 { log.ErrorByKv("get friend list rpc server failed", req.OperationID, "err", err.Error(), "req", req.String()) + + } + self, err := im_mysql_model.FindUserByUID(claims.UID) + if err != nil { + log.ErrorByKv("get self info failed", req.OperationID, "err", err.Error(), "req", req.String()) + } + var name, faceUrl string + if self != nil { + name, faceUrl = self.Name, self.Icon } for _, v := range RpcResp.Data { logic.SendMsgByWS(&pbChat.WSToMsgSvrChatMsg{ - SendID: claims.UID, - RecvID: v.Uid, - Content: claims.UID + "'s info has changed", - SendTime: utils.GetCurrentTimestampBySecond(), - MsgFrom: constant.SysMsgType, - ContentType: constant.SetSelfInfoTip, - SessionType: constant.SingleChatType, - OperationID: req.OperationID, - Token: req.Token, + SendID: claims.UID, + RecvID: v.Uid, + SenderNickName: name, + SenderFaceURL: faceUrl, + Content: claims.UID + "'s info has changed", + SendTime: utils.GetCurrentTimestampBySecond(), + MsgFrom: constant.SysMsgType, + ContentType: constant.SetSelfInfoTip, + SessionType: constant.SingleChatType, + OperationID: req.OperationID, + Token: req.Token, }) }