Compare commits

...

8 Commits

Author SHA1 Message Date
boyce
6c44ba180c 优化kafkamoudule日志 2025-05-21 21:38:36 +08:00
boyce
ecfd42bdec 优化消息队列日志 2025-05-21 21:34:32 +08:00
boyce
01d3b3e535 优化RankService日志 2025-05-21 21:22:24 +08:00
boyce
550d65a354 优化mysql模块日志 2025-05-21 21:19:03 +08:00
boyce
15580ffce9 新增wss证书配置支持 2025-04-21 21:23:34 +08:00
boyce
bd467a219b 废弃掉HttpService、TcpService、WSService 2025-03-28 10:33:51 +08:00
boyce
af15615345 优化日志生成路径 2025-03-14 18:03:01 +08:00
boyce
50dd80b082 整理代码 2025-03-10 11:35:19 +08:00
20 changed files with 101 additions and 88 deletions

View File

@@ -40,8 +40,6 @@ type NodeInfo struct {
DiscoveryService []DiscoveryService //筛选发现的服务,如果不配置,不进行筛选 DiscoveryService []DiscoveryService //筛选发现的服务,如果不配置,不进行筛选
status NodeStatus status NodeStatus
Retire bool Retire bool
NetworkName string
} }
type NodeRpcInfo struct { type NodeRpcInfo struct {

View File

@@ -3,24 +3,23 @@ package cluster
import "github.com/duanhf2012/origin/v2/rpc" import "github.com/duanhf2012/origin/v2/rpc"
type ConfigDiscovery struct { type ConfigDiscovery struct {
funDelNode FunDelNode funDelNode FunDelNode
funSetNode FunSetNode funSetNode FunSetNode
localNodeId string localNodeId string
} }
func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string,funDelNode FunDelNode,funSetNode FunSetNode) error{
discovery.localNodeId = localNodeId discovery.localNodeId = localNodeId
discovery.funDelNode = funDelNode discovery.funDelNode = funDelNode
discovery.funSetNode = funSetNode discovery.funSetNode = funSetNode
//解析本地其他服务配置 //解析本地其他服务配置
_,nodeInfoList,_,err := GetCluster().readLocalClusterConfig(rpc.NodeIdNull) _, nodeInfoList, _, err := GetCluster().readLocalClusterConfig(rpc.NodeIdNull)
if err != nil { if err != nil {
return err return err
} }
for _,nodeInfo := range nodeInfoList { for _, nodeInfo := range nodeInfoList {
if nodeInfo.NodeId == localNodeId { if nodeInfo.NodeId == localNodeId {
continue continue
} }
@@ -30,5 +29,3 @@ func (discovery *ConfigDiscovery) InitDiscovery(localNodeId string,funDelNode Fu
return nil return nil
} }

View File

@@ -5,17 +5,17 @@ import (
"github.com/duanhf2012/origin/v2/service" "github.com/duanhf2012/origin/v2/service"
) )
func (cls *Cluster) setupDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{ func (cls *Cluster) setupDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.discoveryInfo.getDiscoveryType() == OriginType { //origin类型服务发现 if cls.discoveryInfo.getDiscoveryType() == OriginType { //origin类型服务发现
return cls.setupOriginDiscovery(localNodeId,setupServiceFun) return cls.setupOriginDiscovery(localNodeId, setupServiceFun)
}else if cls.discoveryInfo.getDiscoveryType() == EtcdType{//etcd类型服务发现 } else if cls.discoveryInfo.getDiscoveryType() == EtcdType { //etcd类型服务发现
return cls.setupEtcdDiscovery(localNodeId,setupServiceFun) return cls.setupEtcdDiscovery(localNodeId, setupServiceFun)
} }
return cls.setupConfigDiscovery(localNodeId,setupServiceFun) return cls.setupConfigDiscovery(localNodeId, setupServiceFun)
} }
func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{ func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil { if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup") return errors.New("service discovery has been setup")
} }
@@ -27,6 +27,7 @@ func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun Set
} }
cls.serviceDiscovery = getOriginDiscovery() cls.serviceDiscovery = getOriginDiscovery()
//2.如果为动态服务发现安装本地发现服务 //2.如果为动态服务发现安装本地发现服务
if localMaster == true { if localMaster == true {
setupServiceFun(&masterService) setupServiceFun(&masterService)
@@ -36,11 +37,10 @@ func (cls *Cluster) setupOriginDiscovery(localNodeId string, setupServiceFun Set
setupServiceFun(&clientService) setupServiceFun(&clientService)
cls.AddDiscoveryService(OriginDiscoveryClientName, true) cls.AddDiscoveryService(OriginDiscoveryClientName, true)
return nil return nil
} }
func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{ func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil { if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup") return errors.New("service discovery has been setup")
} }
@@ -48,12 +48,12 @@ func (cls *Cluster) setupEtcdDiscovery(localNodeId string, setupServiceFun Setup
//setup etcd service //setup etcd service
cls.serviceDiscovery = getEtcdDiscovery() cls.serviceDiscovery = getEtcdDiscovery()
setupServiceFun(cls.serviceDiscovery.(service.IService)) setupServiceFun(cls.serviceDiscovery.(service.IService))
cls.AddDiscoveryService(cls.serviceDiscovery.(service.IService).GetName(),false) cls.AddDiscoveryService(cls.serviceDiscovery.(service.IService).GetName(), false)
return nil return nil
} }
func (cls *Cluster) setupConfigDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error{ func (cls *Cluster) setupConfigDiscovery(localNodeId string, setupServiceFun SetupServiceFun) error {
if cls.serviceDiscovery != nil { if cls.serviceDiscovery != nil {
return errors.New("service discovery has been setup") return errors.New("service discovery has been setup")
} }

View File

@@ -1,6 +1,11 @@
package cluster package cluster
import ( import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"github.com/duanhf2012/origin/v2/event" "github.com/duanhf2012/origin/v2/event"
"github.com/duanhf2012/origin/v2/log" "github.com/duanhf2012/origin/v2/log"
"github.com/duanhf2012/origin/v2/rpc" "github.com/duanhf2012/origin/v2/rpc"
@@ -9,16 +14,11 @@ import (
"go.etcd.io/etcd/api/v3/mvccpb" "go.etcd.io/etcd/api/v3/mvccpb"
"go.etcd.io/etcd/client/v3" "go.etcd.io/etcd/client/v3"
"google.golang.org/protobuf/proto" "google.golang.org/protobuf/proto"
"time" "os"
"context"
"errors"
"fmt"
"path" "path"
"strings" "strings"
"sync/atomic" "sync/atomic"
"io/ioutil" "time"
"crypto/x509"
"crypto/tls"
) )
const originDir = "/origin" const originDir = "/origin"
@@ -42,7 +42,8 @@ type EtcdDiscoveryService struct {
mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId] mapDiscoveryNodeId map[string]map[string]struct{} //map[networkName]map[nodeId]
} }
var etcdDiscovery *EtcdDiscoveryService var etcdDiscovery *EtcdDiscoveryService
func getEtcdDiscovery() IServiceDiscovery { func getEtcdDiscovery() IServiceDiscovery {
if etcdDiscovery == nil { if etcdDiscovery == nil {
etcdDiscovery = &EtcdDiscoveryService{} etcdDiscovery = &EtcdDiscoveryService{}
@@ -51,7 +52,6 @@ func getEtcdDiscovery() IServiceDiscovery {
return etcdDiscovery return etcdDiscovery
} }
func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error { func (ed *EtcdDiscoveryService) InitDiscovery(localNodeId string, funDelNode FunDelNode, funSetNode FunSetNode) error {
ed.localNodeId = localNodeId ed.localNodeId = localNodeId
@@ -99,17 +99,17 @@ func (ed *EtcdDiscoveryService) OnInit() error {
if etcdDiscoveryCfg.EtcdList[i].Cert != "" { if etcdDiscoveryCfg.EtcdList[i].Cert != "" {
// load cert // load cert
cert, cerr := tls.LoadX509KeyPair(etcdDiscoveryCfg.EtcdList[i].Cert, etcdDiscoveryCfg.EtcdList[i].CertKey) cert, cErr := tls.LoadX509KeyPair(etcdDiscoveryCfg.EtcdList[i].Cert, etcdDiscoveryCfg.EtcdList[i].CertKey)
if cerr != nil { if cErr != nil {
log.Error("load cert error", log.ErrorField("err", cerr)) log.Error("load cert error", log.ErrorField("err", cErr))
return cerr return cErr
} }
// load root ca // load root ca
caData, cerr := ioutil.ReadFile(etcdDiscoveryCfg.EtcdList[i].Ca) caData, cErr := os.ReadFile(etcdDiscoveryCfg.EtcdList[i].Ca)
if cerr != nil { if cErr != nil {
log.Error("load root ca error", log.ErrorField("err", cerr)) log.Error("load root ca error", log.ErrorField("err", cErr))
return cerr return cErr
} }
pool := x509.NewCertPool() pool := x509.NewCertPool()
pool.AppendCertsFromPEM(caData) pool.AppendCertsFromPEM(caData)
@@ -122,13 +122,12 @@ func (ed *EtcdDiscoveryService) OnInit() error {
client, err = clientv3.New(clientv3.Config{ client, err = clientv3.New(clientv3.Config{
Endpoints: etcdDiscoveryCfg.EtcdList[i].Endpoints, Endpoints: etcdDiscoveryCfg.EtcdList[i].Endpoints,
DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond, DialTimeout: etcdDiscoveryCfg.DialTimeoutMillisecond,
Username: etcdDiscoveryCfg.EtcdList[i].UserName, Username: etcdDiscoveryCfg.EtcdList[i].UserName,
Password: etcdDiscoveryCfg.EtcdList[i].Password, Password: etcdDiscoveryCfg.EtcdList[i].Password,
Logger: log.GetLogger().Logger, Logger: log.GetLogger().Logger,
TLS: tlsConfig, TLS: tlsConfig,
}) })
if err != nil { if err != nil {
log.Error("etcd discovery init fail", log.ErrorField("err", err)) log.Error("etcd discovery init fail", log.ErrorField("err", err))
return err return err

View File

@@ -73,9 +73,22 @@ func (handler *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
handler.conns[conn] = struct{}{} handler.conns[conn] = struct{}{}
handler.mutexConns.Unlock() handler.mutexConns.Unlock()
c,ok:=conn.NetConn().(*net.TCPConn)
if !ok {
tlsConn,ok := conn.NetConn().(*tls.Conn)
if !ok {
log.Error("conn error")
return
}
c,ok = tlsConn.NetConn().(*net.TCPConn)
if !ok {
log.Error("conn error")
return
}
}
conn.UnderlyingConn().(*net.TCPConn).SetLinger(0) c.SetLinger(0)
conn.UnderlyingConn().(*net.TCPConn).SetNoDelay(true) c.SetNoDelay(true)
wsConn := newWSConn(conn, r.Header, handler.pendingWriteNum, handler.maxMsgLen, handler.messageType) wsConn := newWSConn(conn, r.Header, handler.pendingWriteNum, handler.maxMsgLen, handler.messageType)
agent := handler.newAgent(wsConn) agent := handler.newAgent(wsConn)
agent.Run() agent.Run()

View File

@@ -479,11 +479,7 @@ func setLogPath(args interface{}) error {
return nil return nil
} }
logPath := strings.TrimSpace(args.(string)) logPath := strings.TrimSpace(args.(string))
dir, err := os.Stat(logPath) _, err := os.Stat(logPath)
if err != nil || dir.IsDir() == false {
return errors.New("Not found dir " + logPath)
}
if err != nil { if err != nil {
err = os.MkdirAll(logPath, os.ModePerm) err = os.MkdirAll(logPath, os.ModePerm)
if err != nil { if err != nil {

View File

@@ -192,7 +192,7 @@ func (s *Service) run() {
break break
} }
if s.profiler != nil { if s.profiler != nil {
analyzer = s.profiler.Push("[Req]" + rpcRequest.RpcRequestData.GetServiceMethod()) analyzer = s.profiler.Push("[RpcReq]" + rpcRequest.RpcRequestData.GetServiceMethod()+"."+strconv.Itoa(int(rpcRequest.RpcRequestData.GetRpcMethodId())))
} }
s.GetRpcHandler().HandlerRpcRequest(rpcRequest) s.GetRpcHandler().HandlerRpcRequest(rpcRequest)

View File

@@ -248,7 +248,7 @@ func (ch *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession
select { select {
case msg := <-claim.Messages(): case msg := <-claim.Messages():
if msg == nil { if msg == nil {
log.SWarning("claim will exit", log.Any("topic", claim.Topic()), log.Any("Partition", claim.Partition())) log.SWarn("claim will exit", log.Any("topic", claim.Topic()), log.Any("Partition", claim.Partition()))
return nil return nil
} }
ch.AppendMsg(session, msg) ch.AppendMsg(session, msg)

View File

@@ -86,7 +86,7 @@ func (p *Producer) asyncRun() {
asyncReturn := sm.Metadata.(*AsyncReturn) asyncReturn := sm.Metadata.(*AsyncReturn)
asyncReturn.chanReturn <- asyncReturn asyncReturn.chanReturn <- asyncReturn
case em := <-p.Errors(): case em := <-p.Errors():
log.Error("async kafkamodule error", log.ErrorAttr("err", em.Err)) log.Error("async kafkamodule error", log.ErrorField("err", em.Err))
if em.Msg.Metadata == nil { if em.Msg.Metadata == nil {
break break
} }

View File

@@ -95,7 +95,7 @@ func (m *MySQLModule) Begin() (*Tx, error) {
var txDBModule Tx var txDBModule Tx
txDb, err := m.db.Begin() txDb, err := m.db.Begin()
if err != nil { if err != nil {
log.Error("Begin error:%s", err.Error()) log.Error("Begin error", log.ErrorField("err",err))
return &txDBModule, err return &txDBModule, err
} }
txDBModule.slowDuration = m.slowDuration txDBModule.slowDuration = m.slowDuration
@@ -155,7 +155,7 @@ func (m *MySQLModule) runPing() {
for { for {
select { select {
case <-m.pingCoroutine.pintExit: case <-m.pingCoroutine.pintExit:
log.Error("RunPing stopping %s...", fmt.Sprintf("%T", m)) log.Error("RunPing stopping",log.String("url", m.url),log.String("dbname", m.dbname))
return return
case <-m.pingCoroutine.tickerPing.C: case <-m.pingCoroutine.tickerPing.C:
if m.db != nil { if m.db != nil {
@@ -221,12 +221,12 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
datasetList.blur = true datasetList.blur = true
if checkArgs(args) != nil { if checkArgs(args) != nil {
log.Error("CheckArgs is error :%s", strQuery) log.Error("CheckArgs is error",log.String("sql",strQuery))
return &datasetList, fmt.Errorf("checkArgs is error") return &datasetList, fmt.Errorf("checkArgs is error")
} }
if db == nil { if db == nil {
log.Error("cannot connect database:%s", strQuery) log.Error("cannot connect database",log.String("sql", strQuery))
return &datasetList, fmt.Errorf("cannot connect database") return &datasetList, fmt.Errorf("cannot connect database")
} }
@@ -235,10 +235,10 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
timeFuncPass := time.Since(TimeFuncStart) timeFuncPass := time.Since(TimeFuncStart)
if checkSlow(slowDuration, timeFuncPass) { if checkSlow(slowDuration, timeFuncPass) {
log.Error("DBModule QueryEx Time %s , Query :%s , args :%+v", timeFuncPass, strQuery, args) log.Error("Query slow",log.Int64("time_ms",timeFuncPass.Milliseconds()),log.String("sql", strQuery), log.Any("args",args))
} }
if err != nil { if err != nil {
log.Error("Query:%s(%v)", strQuery, err) log.Error("Query error", log.String("sql",strQuery),log.ErrorField("err",err))
if rows != nil { if rows != nil {
rows.Close() rows.Close()
} }
@@ -278,8 +278,8 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
hasRet := rows.NextResultSet() hasRet := rows.NextResultSet()
if hasRet == false { if hasRet == false {
if rows.Err() != nil { if rowErr :=rows.Err();rowErr != nil {
log.Error("Query:%s(%+v)", strQuery, rows) log.Error("NextResultSet error", log.String("sql",strQuery), log.ErrorField("err",rowErr))
} }
break break
} }
@@ -291,12 +291,12 @@ func query(slowDuration time.Duration, db dbControl, strQuery string, args ...in
func exec(slowDuration time.Duration, db dbControl, strSql string, args ...interface{}) (*DBResult, error) { func exec(slowDuration time.Duration, db dbControl, strSql string, args ...interface{}) (*DBResult, error) {
ret := &DBResult{} ret := &DBResult{}
if db == nil { if db == nil {
log.Error("cannot connect database:%s", strSql) log.Error("cannot connect database", log.String("sql",strSql))
return ret, fmt.Errorf("cannot connect database") return ret, fmt.Errorf("cannot connect database")
} }
if checkArgs(args) != nil { if checkArgs(args) != nil {
log.Error("CheckArgs is error :%s", strSql) log.Error("CheckArgs is error", log.String("sql",strSql))
return ret, fmt.Errorf("checkArgs is error") return ret, fmt.Errorf("checkArgs is error")
} }
@@ -304,10 +304,10 @@ func exec(slowDuration time.Duration, db dbControl, strSql string, args ...inter
res, err := db.Exec(strSql, args...) res, err := db.Exec(strSql, args...)
timeFuncPass := time.Since(TimeFuncStart) timeFuncPass := time.Since(TimeFuncStart)
if checkSlow(slowDuration, timeFuncPass) { if checkSlow(slowDuration, timeFuncPass) {
log.Error("DBModule QueryEx Time %s , Query :%s , args :%+v", timeFuncPass, strSql, args) log.Error("Exec slow",log.Int64("time_ms",timeFuncPass.Milliseconds()),log.String("sql",strSql),log.Any("args",args) )
} }
if err != nil { if err != nil {
log.Error("Exec:%s(%v)", strSql, err) log.Error("Exec error",log.String("sql",strSql),log.ErrorField("err", err))
return nil, err return nil, err
} }

View File

@@ -34,6 +34,8 @@ type WSCfg struct {
PendingWriteNum int PendingWriteNum int
MaxMsgLen uint32 MaxMsgLen uint32
LittleEndian bool //是否小端序 LittleEndian bool //是否小端序
KeyFile string
CertFile string
} }
type WSPackType int8 type WSPackType int8
@@ -62,13 +64,18 @@ func (ws *WSModule) OnInit() error {
ws.WSServer.MaxMsgLen = ws.wsCfg.MaxMsgLen ws.WSServer.MaxMsgLen = ws.wsCfg.MaxMsgLen
ws.WSServer.Addr = ws.wsCfg.ListenAddr ws.WSServer.Addr = ws.wsCfg.ListenAddr
//3.设置解析处理器 if ws.wsCfg.KeyFile != "" && ws.wsCfg.CertFile != "" {
ws.WSServer.KeyFile = ws.wsCfg.KeyFile
ws.WSServer.CertFile = ws.wsCfg.CertFile
}
// 设置解析处理器
ws.process.SetByteOrder(ws.wsCfg.LittleEndian) ws.process.SetByteOrder(ws.wsCfg.LittleEndian)
ws.mapClient = make(map[string]*WSClient, ws.WSServer.MaxConnNum) ws.mapClient = make(map[string]*WSClient, ws.WSServer.MaxConnNum)
ws.WSServer.NewAgent = ws.NewWSClient ws.WSServer.NewAgent = ws.NewWSClient
//4.设置网络事件处理 // 设置网络事件处理
ws.GetEventProcessor().RegEventReceiverFunc(event.Sys_Event_WebSocket, ws.GetEventHandler(), ws.wsEventHandler) ws.GetEventProcessor().RegEventReceiverFunc(event.Sys_Event_WebSocket, ws.GetEventHandler(), ws.wsEventHandler)
return nil return nil

View File

@@ -83,6 +83,7 @@ type HttpSession struct {
sessionDone chan *HttpSession sessionDone chan *HttpSession
} }
// Deprecated: replace it with the GinModule
type HttpService struct { type HttpService struct {
service.Service service.Service

View File

@@ -104,11 +104,11 @@ func (cs *CustomerSubscriber) UnSubscribe() {
func (cs *CustomerSubscriber) LoadLastIndex() { func (cs *CustomerSubscriber) LoadLastIndex() {
for { for {
if atomic.LoadInt32(&cs.isStop) != 0 { if atomic.LoadInt32(&cs.isStop) != 0 {
log.Info("topic ", cs.topic, " out of subscription") log.SInfo("topic ", cs.topic, " out of subscription")
break break
} }
log.Info("customer ", cs.customerId, " start load last index ") log.SInfo("customer ", cs.customerId, " start load last index ")
lastIndex, ret := cs.subscriber.dataPersist.LoadCustomerIndex(cs.topic, cs.customerId) lastIndex, ret := cs.subscriber.dataPersist.LoadCustomerIndex(cs.topic, cs.customerId)
if ret == true { if ret == true {
if lastIndex > 0 { if lastIndex > 0 {
@@ -116,18 +116,18 @@ func (cs *CustomerSubscriber) LoadLastIndex() {
} else { } else {
//否则直接使用客户端发回来的 //否则直接使用客户端发回来的
} }
log.Info("customer ", cs.customerId, " load finish,start index is ", cs.StartIndex) log.SInfo("customer ", cs.customerId, " load finish,start index is ", cs.StartIndex)
break break
} }
log.Info("customer ", cs.customerId, " load last index is fail...") log.SInfo("customer ", cs.customerId, " load last index is fail...")
time.Sleep(5 * time.Second) time.Sleep(5 * time.Second)
} }
} }
func (cs *CustomerSubscriber) SubscribeRun() { func (cs *CustomerSubscriber) SubscribeRun() {
defer cs.subscriber.queueWait.Done() defer cs.subscriber.queueWait.Done()
log.Info("topic ", cs.topic, " start subscription") log.SInfo("topic ", cs.topic, " start subscription")
//加载之前的位置 //加载之前的位置
if cs.subscribeMethod == MethodLast { if cs.subscribeMethod == MethodLast {
@@ -136,7 +136,7 @@ func (cs *CustomerSubscriber) SubscribeRun() {
for { for {
if atomic.LoadInt32(&cs.isStop) != 0 { if atomic.LoadInt32(&cs.isStop) != 0 {
log.Info("topic ", cs.topic, " out of subscription") log.SInfo("topic ", cs.topic, " out of subscription")
break break
} }
@@ -146,14 +146,14 @@ func (cs *CustomerSubscriber) SubscribeRun() {
//todo 检测退出 //todo 检测退出
if cs.subscribe() == false { if cs.subscribe() == false {
log.Info("topic ", cs.topic, " out of subscription") log.SInfo("topic ", cs.topic, " out of subscription")
break break
} }
} }
//删除订阅关系 //删除订阅关系
cs.subscriber.removeCustomer(cs.customerId, cs) cs.subscriber.removeCustomer(cs.customerId, cs)
log.Info("topic ", cs.topic, " unsubscription") log.SInfo("topic ", cs.topic, " unsubscription")
} }
func (cs *CustomerSubscriber) subscribe() bool { func (cs *CustomerSubscriber) subscribe() bool {

View File

@@ -63,7 +63,7 @@ func (ms *MessageQueueService) ReadCfg() error {
maxProcessTopicBacklogNum, ok := mapDBServiceCfg["MaxProcessTopicBacklogNum"] maxProcessTopicBacklogNum, ok := mapDBServiceCfg["MaxProcessTopicBacklogNum"]
if ok == false { if ok == false {
ms.maxProcessTopicBacklogNum = DefaultMaxTopicBacklogNum ms.maxProcessTopicBacklogNum = DefaultMaxTopicBacklogNum
log.Info("MaxProcessTopicBacklogNum config is set to the default value of ", maxProcessTopicBacklogNum) log.SInfo("MaxProcessTopicBacklogNum config is set to the default value of ", maxProcessTopicBacklogNum)
} else { } else {
ms.maxProcessTopicBacklogNum = int32(maxProcessTopicBacklogNum.(float64)) ms.maxProcessTopicBacklogNum = int32(maxProcessTopicBacklogNum.(float64))
} }
@@ -71,7 +71,7 @@ func (ms *MessageQueueService) ReadCfg() error {
memoryQueueLen, ok := mapDBServiceCfg["MemoryQueueLen"] memoryQueueLen, ok := mapDBServiceCfg["MemoryQueueLen"]
if ok == false { if ok == false {
ms.memoryQueueLen = DefaultMemoryQueueLen ms.memoryQueueLen = DefaultMemoryQueueLen
log.Info("MemoryQueueLen config is set to the default value of ", DefaultMemoryQueueLen) log.SInfo("MemoryQueueLen config is set to the default value of ", DefaultMemoryQueueLen)
} else { } else {
ms.memoryQueueLen = int32(memoryQueueLen.(float64)) ms.memoryQueueLen = int32(memoryQueueLen.(float64))
} }

View File

@@ -237,7 +237,7 @@ func (mp *MongoPersist) findTopicData(topic string, startIndex uint64, limit int
defer cancelAll() defer cancelAll()
err = cursor.All(ctxAll, &res) err = cursor.All(ctxAll, &res)
if err != nil { if err != nil {
log.Error("find collect name ", topic, " is error", log.ErrorAttr("err", err)) log.Error("find collect name error",log.String("topic",topic) ,log.ErrorField("err",err))
return nil, false return nil, false
} }
@@ -246,7 +246,7 @@ func (mp *MongoPersist) findTopicData(topic string, startIndex uint64, limit int
rawData, errM := bson.Marshal(res[i]) rawData, errM := bson.Marshal(res[i])
if errM != nil { if errM != nil {
if errM != nil { if errM != nil {
log.Error("collect name ", topic, " Marshal is error", log.ErrorAttr("err", err)) log.Error("Marshal error",log.String("topic",topic) , log.ErrorField("err", err))
return nil, false return nil, false
} }
continue continue
@@ -391,7 +391,7 @@ func (mp *MongoPersist) GetIndex(topicData *TopicData) uint64 {
if e.Key == "_id" { if e.Key == "_id" {
errC, seq := convertToNumber[uint64](e.Value) errC, seq := convertToNumber[uint64](e.Value)
if errC != nil { if errC != nil {
log.Error("value is error:%s,%+v, ", errC.Error(), e.Value) log.Error("value is error", log.ErrorField("err",errC), log.Any("val",e.Value))
} }
return seq return seq

View File

@@ -56,9 +56,9 @@ func (ss *Subscriber) TopicSubscribe(rpcHandler rpc.IRpcHandler, subScribeType r
} }
if ok == true { if ok == true {
log.Info("repeat subscription for customer ", customerId) log.SInfo("repeat subscription for customer ", customerId)
} else { } else {
log.Info("subscription for customer ", customerId) log.SInfo("subscription for customer ", customerId)
} }
} }
@@ -72,7 +72,7 @@ func (ss *Subscriber) UnSubscribe(customerId string) {
customerSubscriber, ok := ss.mapCustomer[customerId] customerSubscriber, ok := ss.mapCustomer[customerId]
if ok == false { if ok == false {
log.SWarning("failed to unsubscribe customer " + customerId) log.SWarn("failed to unsubscribe customer ", customerId)
return return
} }

View File

@@ -93,7 +93,7 @@ func (tr *TopicRoom) Stop() {
func (tr *TopicRoom) topicRoomRun() { func (tr *TopicRoom) topicRoomRun() {
defer tr.queueWait.Done() defer tr.queueWait.Done()
log.Info("topic room ", tr.topic, " is running..") log.SInfo("topic room ", tr.topic, " is running..")
for { for {
if atomic.LoadInt32(&tr.isStop) != 0 { if atomic.LoadInt32(&tr.isStop) != 0 {
break break
@@ -145,5 +145,5 @@ func (tr *TopicRoom) topicRoomRun() {
} }
tr.customerLocker.Unlock() tr.customerLocker.Unlock()
log.Info("topic room ", tr.topic, " is stop") log.SInfo("topic room ", tr.topic, " is stop")
} }

View File

@@ -142,13 +142,13 @@ func (mp *MongoPersist) OnSetupRank(manual bool, rankSkip *RankSkip) error {
return nil return nil
} }
log.Info("start load rank ", rankSkip.GetRankName(), " from mongodb.") log.SInfo("start load rank ", rankSkip.GetRankName(), " from mongodb.")
err := mp.loadFromDB(rankSkip.GetRankID(), rankSkip.GetRankName()) err := mp.loadFromDB(rankSkip.GetRankID(), rankSkip.GetRankName())
if err != nil { if err != nil {
log.SError("load from db is fail :%s", err.Error()) log.SError("load from db is fail :%s", err.Error())
return err return err
} }
log.Info("finish load rank ", rankSkip.GetRankName(), " from mongodb.") log.SInfo("finish load rank ", rankSkip.GetRankName(), " from mongodb.")
return nil return nil
} }
@@ -296,7 +296,7 @@ func (mp *MongoPersist) saveToDB() {
buf := make([]byte, 4096) buf := make([]byte, 4096)
l := runtime.Stack(buf, false) l := runtime.Stack(buf, false)
errString := fmt.Sprint(r) errString := fmt.Sprint(r)
log.Dump(string(buf[:l]), log.String("error", errString)) log.StackError(string(buf[:l]), log.String("error", errString))
} }
}() }()

View File

@@ -14,6 +14,7 @@ import (
"time" "time"
) )
// Deprecated: replace it with the TcpModule
type TcpService struct { type TcpService struct {
tcpServer network.TCPServer tcpServer network.TCPServer
service.Service service.Service

View File

@@ -12,6 +12,7 @@ import (
"sync" "sync"
) )
// Deprecated: replace it with the WSModule
type WSService struct { type WSService struct {
service.Service service.Service
wsServer network.WSServer wsServer network.WSServer