blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-06 04:15:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

test recover

Browse Source
This commit is contained in:
Gordon 2022-03-04 16:11:04 +08:00
parent c108115c83
commit 8dc04667b0
4 changed files with 80 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
internal/msg_gateway/gate/ws_server.go
View File

@ -5,6 +5,7 @@ import (
"Open_IM/pkg/common/constant" "Open_IM/pkg/common/constant"
"Open_IM/pkg/common/db" "Open_IM/pkg/common/db"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
"Open_IM/pkg/common/token_verify"
"Open_IM/pkg/utils" "Open_IM/pkg/utils"
"bytes" "bytes"
"encoding/gob" "encoding/gob"
@ -257,16 +258,16 @@ func (ws *WServer) headerCheck(w http.ResponseWriter, r *http.Request) bool {
status := http.StatusUnauthorized status := http.StatusUnauthorized
query := r.URL.Query() query := r.URL.Query()
if len(query["token"]) != 0 && len(query["sendID"]) != 0 && len(query["platformID"]) != 0 { if len(query["token"]) != 0 && len(query["sendID"]) != 0 && len(query["platformID"]) != 0 {
//if ok, err := token_verify.VerifyToken(query["token"][0], query["sendID"][0]); !ok { if ok, err := token_verify.VerifyToken(query["token"][0], query["sendID"][0]); !ok {
// e := err.(*constant.ErrInfo) e := err.(*constant.ErrInfo)
// log.ErrorByKv("Token verify failed", "", "query", query) log.ErrorByKv("Token verify failed", "", "query", query)
// w.Header().Set("Sec-Websocket-Version", "13") w.Header().Set("Sec-Websocket-Version", "13")
// http.Error(w, e.ErrMsg, int(e.ErrCode)) http.Error(w, e.ErrMsg, int(e.ErrCode))
// return false return false
//} else { } else {
log.InfoByKv("Connection Authentication Success", "", "token", query["token"][0], "userID", query["sendID"][0]) log.InfoByKv("Connection Authentication Success", "", "token", query["token"][0], "userID", query["sendID"][0])
return true return true
//} }
} else { } else {
log.ErrorByKv("Args err", "", "query", query) log.ErrorByKv("Args err", "", "query", query)
w.Header().Set("Sec-Websocket-Version", "13") w.Header().Set("Sec-Websocket-Version", "13")

2
internal/msg_transfer/logic/init.go
View File

@ -20,6 +20,6 @@ func Init() {
} }
func Run() { func Run() {
//register mysqlConsumerHandler to //register mysqlConsumerHandler to
//go persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(&persistentCH) go persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(&persistentCH)
go historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyCH) go historyCH.historyConsumerGroup.RegisterHandleAndConsumer(&historyCH)
} }

126
internal/push/logic/push_to_client.go
View File

@ -7,12 +7,16 @@
package logic package logic
import ( import (
push "Open_IM/internal/push/jpush"
"Open_IM/pkg/common/config" "Open_IM/pkg/common/config"
"Open_IM/pkg/common/constant"
"Open_IM/pkg/common/log" "Open_IM/pkg/common/log"
"Open_IM/pkg/grpc-etcdv3/getcdv3" "Open_IM/pkg/grpc-etcdv3/getcdv3"
pbPush "Open_IM/pkg/proto/push" pbPush "Open_IM/pkg/proto/push"
pbRelay "Open_IM/pkg/proto/relay" pbRelay "Open_IM/pkg/proto/relay"
"Open_IM/pkg/utils"
"context" "context"
"encoding/json"
"strings" "strings"
) )
@ -30,7 +34,7 @@ type AtContent struct {
func MsgToUser(pushMsg *pbPush.PushMsgReq) { func MsgToUser(pushMsg *pbPush.PushMsgReq) {
var wsResult []*pbRelay.SingleMsgToUser var wsResult []*pbRelay.SingleMsgToUser
//isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush) isOfflinePush := utils.GetSwitchFromOptions(pushMsg.MsgData.Options, constant.IsOfflinePush)
log.Debug("Get msg from msg_transfer And push msg", pushMsg.OperationID, "PushData", pushMsg.String()) log.Debug("Get msg from msg_transfer And push msg", pushMsg.OperationID, "PushData", pushMsg.String())
grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName) grpcCons := getcdv3.GetConn4Unique(config.Config.Etcd.EtcdSchema, strings.Join(config.Config.Etcd.EtcdAddr, ","), config.Config.RpcRegisterName.OpenImOnlineMessageRelayName)
//Online push message //Online push message
@ -48,66 +52,66 @@ func MsgToUser(pushMsg *pbPush.PushMsgReq) {
} }
log.InfoByKv("push_result", pushMsg.OperationID, "result", wsResult, "sendData", pushMsg.MsgData) log.InfoByKv("push_result", pushMsg.OperationID, "result", wsResult, "sendData", pushMsg.MsgData)
count++ count++
//if isOfflinePush && pushMsg.PushToUserID != pushMsg.MsgData.SendID { if isOfflinePush && pushMsg.PushToUserID != pushMsg.MsgData.SendID {
// for _, v := range wsResult { for _, v := range wsResult {
// if v.ResultCode == 0 { if v.ResultCode == 0 {
// continue continue
// } }
// //supported terminal //supported terminal
// for _, t := range pushTerminal { for _, t := range pushTerminal {
// if v.RecvPlatFormID == t { if v.RecvPlatFormID == t {
// //Use offline push messaging //Use offline push messaging
// var UIDList []string var UIDList []string
// UIDList = append(UIDList, v.RecvID) UIDList = append(UIDList, v.RecvID)
// customContent := OpenIMContent{ customContent := OpenIMContent{
// SessionType: int(pushMsg.MsgData.SessionType), SessionType: int(pushMsg.MsgData.SessionType),
// From: pushMsg.MsgData.SendID, From: pushMsg.MsgData.SendID,
// To: pushMsg.MsgData.RecvID, To: pushMsg.MsgData.RecvID,
// Seq: pushMsg.MsgData.Seq, Seq: pushMsg.MsgData.Seq,
// } }
// bCustomContent, _ := json.Marshal(customContent) bCustomContent, _ := json.Marshal(customContent)
// jsonCustomContent := string(bCustomContent) jsonCustomContent := string(bCustomContent)
// var content string var content string
// if pushMsg.MsgData.OfflinePushInfo != nil { if pushMsg.MsgData.OfflinePushInfo != nil {
// content = pushMsg.MsgData.OfflinePushInfo.Title content = pushMsg.MsgData.OfflinePushInfo.Title
//
// } else { } else {
// switch pushMsg.MsgData.ContentType { switch pushMsg.MsgData.ContentType {
// case constant.Text: case constant.Text:
// content = constant.ContentType2PushContent[constant.Text] content = constant.ContentType2PushContent[constant.Text]
// case constant.Picture: case constant.Picture:
// content = constant.ContentType2PushContent[constant.Picture] content = constant.ContentType2PushContent[constant.Picture]
// case constant.Voice: case constant.Voice:
// content = constant.ContentType2PushContent[constant.Voice] content = constant.ContentType2PushContent[constant.Voice]
// case constant.Video: case constant.Video:
// content = constant.ContentType2PushContent[constant.Video] content = constant.ContentType2PushContent[constant.Video]
// case constant.File: case constant.File:
// content = constant.ContentType2PushContent[constant.File] content = constant.ContentType2PushContent[constant.File]
// case constant.AtText: case constant.AtText:
// a := AtContent{} a := AtContent{}
// _ = utils.JsonStringToStruct(string(pushMsg.MsgData.Content), &a) _ = utils.JsonStringToStruct(string(pushMsg.MsgData.Content), &a)
// if utils.IsContain(v.RecvID, a.AtUserList) { if utils.IsContain(v.RecvID, a.AtUserList) {
// content = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common] content = constant.ContentType2PushContent[constant.AtText] + constant.ContentType2PushContent[constant.Common]
// } else { } else {
// content = constant.ContentType2PushContent[constant.GroupMsg] content = constant.ContentType2PushContent[constant.GroupMsg]
// } }
// default: default:
// content = constant.ContentType2PushContent[constant.Common] content = constant.ContentType2PushContent[constant.Common]
// } }
// } }
//
// pushResult, err := push.JGAccountListPush(UIDList, content, jsonCustomContent, constant.PlatformIDToName(t)) pushResult, err := push.JGAccountListPush(UIDList, content, jsonCustomContent, constant.PlatformIDToName(t))
// if err != nil { if err != nil {
// log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error(), constant.PlatformIDToName(t)) log.NewError(pushMsg.OperationID, "offline push error", pushMsg.String(), err.Error(), constant.PlatformIDToName(t))
// } else { } else {
// log.NewDebug(pushMsg.OperationID, "offline push return result is ", string(pushResult), pushMsg.MsgData, constant.PlatformIDToName(t)) log.NewDebug(pushMsg.OperationID, "offline push return result is ", string(pushResult), pushMsg.MsgData, constant.PlatformIDToName(t))
// } }
//
// } }
// } }
// } }
//
//} }
} }
//func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) { //func SendMsgByWS(m *pbChat.WSToMsgSvrChatMsg) {

7
pkg/statistics/statistics.go
View File

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