Compare commits

...

16 Commits

Author SHA1 Message Date
orgin
be3daf19f9 优化服务配置获取函数 2022-06-09 17:44:42 +08:00
orgin
aa91c7bf1b Merge pull request #860 from cgc1983/master
module exit error
2022-06-09 11:52:50 +08:00
cgc1983
7fe73e55fb module exit error 2022-06-09 11:32:25 +08:00
cgc1983
e5ceaa9e76 ignore macosx 2022-06-02 17:51:03 +08:00
orgin
97c55ada71 优化网络连接Id生成规则 2022-06-02 17:07:01 +08:00
orgin
776b234022 优化网络连接Id生成规则&优化WebSocket服务 2022-06-02 16:09:16 +08:00
orgin
a4f425bd69 同步jsonprocessor 2022-05-31 10:57:47 +08:00
orgin
ee54862be2 fixed HTTP status codes write error 2022-05-12 09:45:22 +08:00
orgin
13642d7402 新增name命令参数 2022-04-19 09:26:11 +08:00
orgin
18d620118e 在help中新增编译参数信息 2022-04-13 11:47:13 +08:00
orgin
f02592d126 在help中新增编译参数信息 2022-04-13 11:36:20 +08:00
orgin
0f890887f7 在help中新增编译时间 2022-04-13 10:24:36 +08:00
orgin
c5c05b6ae9 补充Readme:新增Global配置说明 2022-04-02 14:24:46 +08:00
orgin
68b891df51 新增全局配置 2022-04-02 14:02:33 +08:00
orgin
63199bf862 优化mongodbmodule的NextSeq接口 2022-04-01 17:35:13 +08:00
orgin
4d5d45d555 优化mongo模块 2022-04-01 14:06:24 +08:00
15 changed files with 293 additions and 203 deletions

1
.gitignore vendored
View File

@@ -10,3 +10,4 @@
# Output of the go coverage tool, specifically when used with LiteIDE
*.out
.DS_Store

View File

@@ -87,6 +87,9 @@ service.json如下
---------------
```
{
"Global": {
"AreaId": 1
},
"Service":{
"HttpService":{
"ListenAddr":"0.0.0.0:9402",
@@ -157,7 +160,7 @@ service.json如下
```
---------------
以上配置分为两个部分Service与NodeServiceNodeService中配置的对应结点中服务的配置如果启动程序中根据nodeid查找该域的对应的服务如果找不到时从Service公共部分查找。
以上配置分为两个部分:Global,Service与NodeService。Global是全局配置在任何服务中都可以通过cluster.GetCluster().GetGlobalCfg()获取NodeService中配置的对应结点中服务的配置如果启动程序中根据nodeid查找该域的对应的服务如果找不到时从Service公共部分查找。
**HttpService配置**
* ListenAddr:Http监听地址

View File

@@ -41,8 +41,9 @@ type NodeRpcInfo struct {
var cluster Cluster
type Cluster struct {
localNodeInfo NodeInfo //本结点配置信息
masterDiscoveryNodeList []NodeInfo //配置发现Master结点
localNodeInfo NodeInfo //本结点配置信息
masterDiscoveryNodeList []NodeInfo //配置发现Master结点
globalCfg interface{} //全局配置
localServiceCfg map[string]interface{} //map[serviceName]配置数据*
mapRpc map[int]NodeRpcInfo //nodeId
@@ -418,3 +419,7 @@ func HasService(nodeId int, serviceName string) bool {
return false
}
func (cls *Cluster) GetGlobalCfg() interface{} {
return cls.globalCfg
}

View File

@@ -10,12 +10,13 @@ import (
)
var json = jsoniter.ConfigCompatibleWithStandardLibrary
type NodeInfoList struct {
MasterDiscoveryNode []NodeInfo //用于服务发现Node
NodeList []NodeInfo
MasterDiscoveryNode []NodeInfo //用于服务发现Node
NodeList []NodeInfo
}
func (cls *Cluster) ReadClusterConfig(filepath string) (*NodeInfoList,error) {
func (cls *Cluster) ReadClusterConfig(filepath string) (*NodeInfoList, error) {
c := &NodeInfoList{}
d, err := ioutil.ReadFile(filepath)
if err != nil {
@@ -26,119 +27,123 @@ func (cls *Cluster) ReadClusterConfig(filepath string) (*NodeInfoList,error) {
return nil, err
}
return c,nil
return c, nil
}
func (cls *Cluster) readServiceConfig(filepath string) (map[string]interface{},map[int]map[string]interface{},error) {
func (cls *Cluster) readServiceConfig(filepath string) (interface{}, map[string]interface{}, map[int]map[string]interface{}, error) {
c := map[string]interface{}{}
//读取配置
d, err := ioutil.ReadFile(filepath)
if err != nil {
return nil,nil, err
return nil, nil, nil, err
}
err = json.Unmarshal(d, &c)
if err != nil {
return nil,nil, err
return nil, nil, nil, err
}
GlobalCfg, ok := c["Global"]
serviceConfig := map[string]interface{}{}
serviceCfg,ok := c["Service"]
serviceCfg, ok := c["Service"]
if ok == true {
serviceConfig = serviceCfg.(map[string]interface{})
}
mapNodeService := map[int]map[string]interface{}{}
nodeServiceCfg,ok := c["NodeService"]
nodeServiceCfg, ok := c["NodeService"]
if ok == true {
nodeServiceList := nodeServiceCfg.([]interface{})
for _,v := range nodeServiceList{
serviceCfg :=v.(map[string]interface{})
nodeId,ok := serviceCfg["NodeId"]
for _, v := range nodeServiceList {
serviceCfg := v.(map[string]interface{})
nodeId, ok := serviceCfg["NodeId"]
if ok == false {
log.SFatal("NodeService list not find nodeId field")
}
mapNodeService[int(nodeId.(float64))] = serviceCfg
}
}
return serviceConfig,mapNodeService,nil
return GlobalCfg, serviceConfig, mapNodeService, nil
}
func (cls *Cluster) readLocalClusterConfig(nodeId int) ([]NodeInfo,[]NodeInfo,error) {
func (cls *Cluster) readLocalClusterConfig(nodeId int) ([]NodeInfo, []NodeInfo, error) {
var nodeInfoList []NodeInfo
var masterDiscoverNodeList []NodeInfo
clusterCfgPath :=strings.TrimRight(configDir,"/") +"/cluster"
fileInfoList,err := ioutil.ReadDir(clusterCfgPath)
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := ioutil.ReadDir(clusterCfgPath)
if err != nil {
return nil,nil,fmt.Errorf("Read dir %s is fail :%+v",clusterCfgPath,err)
return nil, nil, fmt.Errorf("Read dir %s is fail :%+v", clusterCfgPath, err)
}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _,f := range fileInfoList{
for _, f := range fileInfoList {
if f.IsDir() == false {
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath,"/"),"\\")+"/"+f.Name()
localNodeInfoList,err := cls.ReadClusterConfig(filePath)
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
localNodeInfoList, err := cls.ReadClusterConfig(filePath)
if err != nil {
return nil,nil,fmt.Errorf("read file path %s is error:%+v" ,filePath,err)
return nil, nil, fmt.Errorf("read file path %s is error:%+v", filePath, err)
}
masterDiscoverNodeList = append(masterDiscoverNodeList,localNodeInfoList.MasterDiscoveryNode...)
for _,nodeInfo := range localNodeInfoList.NodeList {
masterDiscoverNodeList = append(masterDiscoverNodeList, localNodeInfoList.MasterDiscoveryNode...)
for _, nodeInfo := range localNodeInfoList.NodeList {
if nodeInfo.NodeId == nodeId || nodeId == 0 {
nodeInfoList = append(nodeInfoList,nodeInfo)
nodeInfoList = append(nodeInfoList, nodeInfo)
}
}
}
}
if nodeId != 0 && (len(nodeInfoList)!=1){
return nil,nil,fmt.Errorf("%d configurations were found for the configuration with node ID %d!",len(nodeInfoList),nodeId)
if nodeId != 0 && (len(nodeInfoList) != 1) {
return nil, nil, fmt.Errorf("%d configurations were found for the configuration with node ID %d!", len(nodeInfoList), nodeId)
}
for i,_ := range nodeInfoList{
for j,s := range nodeInfoList[i].ServiceList{
for i, _ := range nodeInfoList {
for j, s := range nodeInfoList[i].ServiceList {
//私有结点不加入到Public服务列表中
if strings.HasPrefix(s,"_") == false && nodeInfoList[i].Private==false {
nodeInfoList[i].PublicServiceList = append(nodeInfoList[i].PublicServiceList,strings.TrimLeft(s,"_"))
}else{
nodeInfoList[i].ServiceList[j] = strings.TrimLeft(s,"_")
if strings.HasPrefix(s, "_") == false && nodeInfoList[i].Private == false {
nodeInfoList[i].PublicServiceList = append(nodeInfoList[i].PublicServiceList, strings.TrimLeft(s, "_"))
} else {
nodeInfoList[i].ServiceList[j] = strings.TrimLeft(s, "_")
}
}
}
return masterDiscoverNodeList,nodeInfoList,nil
return masterDiscoverNodeList, nodeInfoList, nil
}
func (cls *Cluster) readLocalService(localNodeId int) error {
clusterCfgPath :=strings.TrimRight(configDir,"/") +"/cluster"
fileInfoList,err := ioutil.ReadDir(clusterCfgPath)
clusterCfgPath := strings.TrimRight(configDir, "/") + "/cluster"
fileInfoList, err := ioutil.ReadDir(clusterCfgPath)
if err != nil {
return fmt.Errorf("Read dir %s is fail :%+v",clusterCfgPath,err)
return fmt.Errorf("Read dir %s is fail :%+v", clusterCfgPath, err)
}
//读取任何文件,只读符合格式的配置,目录下的文件可以自定义分文件
for _,f := range fileInfoList {
for _, f := range fileInfoList {
if f.IsDir() == false {
filePath := strings.TrimRight(strings.TrimRight(clusterCfgPath, "/"), "\\") + "/" + f.Name()
serviceConfig,mapNodeService,err := cls.readServiceConfig(filePath)
currGlobalCfg, serviceConfig, mapNodeService, err := cls.readServiceConfig(filePath)
if err != nil {
continue
}
for _,s := range cls.localNodeInfo.ServiceList{
for{
if currGlobalCfg != nil {
cls.globalCfg = currGlobalCfg
}
for _, s := range cls.localNodeInfo.ServiceList {
for {
//取公共服务配置
pubCfg,ok := serviceConfig[s]
pubCfg, ok := serviceConfig[s]
if ok == true {
cls.localServiceCfg[s] = pubCfg
}
//如果结点也配置了该服务,则覆盖之
nodeService,ok := mapNodeService[localNodeId]
nodeService, ok := mapNodeService[localNodeId]
if ok == false {
break
}
sCfg,ok := nodeService[s]
if ok == false{
sCfg, ok := nodeService[s]
if ok == false {
break
}
@@ -152,22 +157,21 @@ func (cls *Cluster) readLocalService(localNodeId int) error {
return nil
}
func (cls *Cluster) parseLocalCfg(){
func (cls *Cluster) parseLocalCfg() {
cls.mapIdNode[cls.localNodeInfo.NodeId] = cls.localNodeInfo
for _,sName := range cls.localNodeInfo.ServiceList{
if _,ok:=cls.mapServiceNode[sName];ok==false{
for _, sName := range cls.localNodeInfo.ServiceList {
if _, ok := cls.mapServiceNode[sName]; ok == false {
cls.mapServiceNode[sName] = make(map[int]struct{})
}
cls.mapServiceNode[sName][cls.localNodeInfo.NodeId]= struct{}{}
cls.mapServiceNode[sName][cls.localNodeInfo.NodeId] = struct{}{}
}
}
func (cls *Cluster) checkDiscoveryNodeList(discoverMasterNode []NodeInfo) bool{
for i:=0;i<len(discoverMasterNode)-1;i++{
for j:=i+1;j<len(discoverMasterNode);j++{
func (cls *Cluster) checkDiscoveryNodeList(discoverMasterNode []NodeInfo) bool {
for i := 0; i < len(discoverMasterNode)-1; i++ {
for j := i + 1; j < len(discoverMasterNode); j++ {
if discoverMasterNode[i].NodeId == discoverMasterNode[j].NodeId ||
discoverMasterNode[i].ListenAddr == discoverMasterNode[j].ListenAddr {
return false
@@ -178,19 +182,19 @@ func (cls *Cluster) checkDiscoveryNodeList(discoverMasterNode []NodeInfo) bool{
return true
}
func (cls *Cluster) InitCfg(localNodeId int) error{
func (cls *Cluster) InitCfg(localNodeId int) error {
cls.localServiceCfg = map[string]interface{}{}
cls.mapRpc = map[int] NodeRpcInfo{}
cls.mapRpc = map[int]NodeRpcInfo{}
cls.mapIdNode = map[int]NodeInfo{}
cls.mapServiceNode = map[string]map[int]struct{}{}
//加载本地结点的NodeList配置
discoveryNode,nodeInfoList,err := cls.readLocalClusterConfig(localNodeId)
discoveryNode, nodeInfoList, err := cls.readLocalClusterConfig(localNodeId)
if err != nil {
return err
}
cls.localNodeInfo = nodeInfoList[0]
if cls.checkDiscoveryNodeList(discoveryNode) ==false {
if cls.checkDiscoveryNodeList(discoveryNode) == false {
return fmt.Errorf("DiscoveryNode config is error!")
}
cls.masterDiscoveryNodeList = discoveryNode
@@ -209,48 +213,39 @@ func (cls *Cluster) InitCfg(localNodeId int) error{
func (cls *Cluster) IsConfigService(serviceName string) bool {
cls.locker.RLock()
defer cls.locker.RUnlock()
mapNode,ok := cls.mapServiceNode[serviceName]
mapNode, ok := cls.mapServiceNode[serviceName]
if ok == false {
return false
}
_,ok = mapNode[cls.localNodeInfo.NodeId]
_, ok = mapNode[cls.localNodeInfo.NodeId]
return ok
}
func (cls *Cluster) GetNodeIdByService(serviceName string,rpcClientList []*rpc.Client,bAll bool) (error,int) {
func (cls *Cluster) GetNodeIdByService(serviceName string, rpcClientList []*rpc.Client, bAll bool) (error, int) {
cls.locker.RLock()
defer cls.locker.RUnlock()
mapNodeId,ok := cls.mapServiceNode[serviceName]
mapNodeId, ok := cls.mapServiceNode[serviceName]
count := 0
if ok == true {
for nodeId,_ := range mapNodeId {
for nodeId, _ := range mapNodeId {
pClient := GetCluster().getRpcClient(nodeId)
if pClient==nil || (bAll == false && pClient.IsConnected()==false) {
if pClient == nil || (bAll == false && pClient.IsConnected() == false) {
continue
}
rpcClientList[count] = pClient
count++
if count>=cap(rpcClientList) {
if count >= cap(rpcClientList) {
break
}
}
}
return nil,count
return nil, count
}
func (cls *Cluster) getServiceCfg(serviceName string) interface{}{
v,ok := cls.localServiceCfg[serviceName]
if ok == false {
return nil
}
return v
}
func (cls *Cluster) GetServiceCfg(serviceName string) interface{}{
serviceCfg,ok := cls.localServiceCfg[serviceName]
func (cls *Cluster) GetServiceCfg(serviceName string) interface{} {
serviceCfg, ok := cls.localServiceCfg[serviceName]
if ok == false {
return nil
}

View File

@@ -5,6 +5,7 @@ import (
"fmt"
"github.com/duanhf2012/origin/network"
"reflect"
"github.com/duanhf2012/origin/log"
)
type MessageJsonInfo struct {
@@ -44,18 +45,18 @@ func (jsonProcessor *JsonProcessor) SetByteOrder(littleEndian bool) {
}
// must goroutine safe
func (jsonProcessor *JsonProcessor ) MsgRoute(msg interface{},userdata interface{}) error{
func (jsonProcessor *JsonProcessor ) MsgRoute(clientId uint64,msg interface{}) error{
pPackInfo := msg.(*JsonPackInfo)
v,ok := jsonProcessor.mapMsg[pPackInfo.typ]
if ok == false {
return fmt.Errorf("cannot find msgtype %d is register!",pPackInfo.typ)
}
v.msgHandler(userdata.(uint64),pPackInfo.msg)
v.msgHandler(clientId,pPackInfo.msg)
return nil
}
func (jsonProcessor *JsonProcessor) Unmarshal(data []byte) (interface{}, error) {
func (jsonProcessor *JsonProcessor) Unmarshal(clientId uint64,data []byte) (interface{}, error) {
typeStruct := struct {Type int `json:"typ"`}{}
defer jsonProcessor.ReleaseByteSlice(data)
err := json.Unmarshal(data, &typeStruct)
@@ -78,7 +79,7 @@ func (jsonProcessor *JsonProcessor) Unmarshal(data []byte) (interface{}, error)
return &JsonPackInfo{typ:msgType,msg:msgData},nil
}
func (jsonProcessor *JsonProcessor) Marshal(msg interface{}) ([]byte, error) {
func (jsonProcessor *JsonProcessor) Marshal(clientId uint64,msg interface{}) ([]byte, error) {
rawMsg,err := json.Marshal(msg)
if err != nil {
return nil,err
@@ -103,16 +104,26 @@ func (jsonProcessor *JsonProcessor) MakeRawMsg(msgType uint16,msg []byte) *JsonP
return &JsonPackInfo{typ:msgType,rawMsg:msg}
}
func (jsonProcessor *JsonProcessor) UnknownMsgRoute(msg interface{}, userData interface{}){
jsonProcessor.unknownMessageHandler(userData.(uint64),msg.([]byte))
func (jsonProcessor *JsonProcessor) UnknownMsgRoute(clientId uint64,msg interface{}){
if jsonProcessor.unknownMessageHandler==nil {
log.SDebug("Unknown message received from ",clientId)
return
}
jsonProcessor.unknownMessageHandler(clientId,msg.([]byte))
}
func (jsonProcessor *JsonProcessor) ConnectedRoute(userData interface{}){
jsonProcessor.connectHandler(userData.(uint64))
func (jsonProcessor *JsonProcessor) ConnectedRoute(clientId uint64){
if jsonProcessor.connectHandler != nil {
jsonProcessor.connectHandler(clientId)
}
}
func (jsonProcessor *JsonProcessor) DisConnectedRoute(userData interface{}){
jsonProcessor.disconnectHandler(userData.(uint64))
func (jsonProcessor *JsonProcessor) DisConnectedRoute(clientId uint64){
if jsonProcessor.disconnectHandler != nil {
jsonProcessor.disconnectHandler(clientId)
}
}
func (jsonProcessor *JsonProcessor) RegisterUnknownMsg(unknownMessageHandler UnknownMessageJsonHandler){

View File

@@ -21,6 +21,7 @@ type WSClient struct {
cons WebsocketConnSet
wg sync.WaitGroup
closeFlag bool
messageType int
}
func (client *WSClient) Start() {
@@ -62,7 +63,7 @@ func (client *WSClient) init() {
if client.cons != nil {
log.SFatal("client is running")
}
client.messageType = websocket.TextMessage
client.cons = make(WebsocketConnSet)
client.closeFlag = false
client.dialer = websocket.Dialer{
@@ -83,6 +84,9 @@ func (client *WSClient) dial() *websocket.Conn {
}
}
func (client *WSClient) SetMessageType(messageType int){
client.messageType = messageType
}
func (client *WSClient) connect() {
defer client.wg.Done()
@@ -102,7 +106,7 @@ reconnect:
client.cons[conn] = struct{}{}
client.Unlock()
wsConn := newWSConn(conn, client.PendingWriteNum, client.MaxMsgLen)
wsConn := newWSConn(conn, client.PendingWriteNum, client.MaxMsgLen,client.messageType)
agent := client.NewAgent(wsConn)
agent.Run()

View File

@@ -18,7 +18,7 @@ type WSConn struct {
closeFlag bool
}
func newWSConn(conn *websocket.Conn, pendingWriteNum int, maxMsgLen uint32) *WSConn {
func newWSConn(conn *websocket.Conn, pendingWriteNum int, maxMsgLen uint32,messageType int) *WSConn {
wsConn := new(WSConn)
wsConn.conn = conn
wsConn.writeChan = make(chan []byte, pendingWriteNum)
@@ -30,7 +30,7 @@ func newWSConn(conn *websocket.Conn, pendingWriteNum int, maxMsgLen uint32) *WSC
break
}
err := conn.WriteMessage(websocket.BinaryMessage, b)
err := conn.WriteMessage(messageType, b)
if err != nil {
break
}

View File

@@ -21,6 +21,7 @@ type WSServer struct {
NewAgent func(*WSConn) Agent
ln net.Listener
handler *WSHandler
messageType int
}
type WSHandler struct {
@@ -32,6 +33,11 @@ type WSHandler struct {
conns WebsocketConnSet
mutexConns sync.Mutex
wg sync.WaitGroup
messageType int
}
func (handler *WSHandler) SetMessageType(messageType int){
handler.messageType = messageType
}
func (handler *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -45,6 +51,7 @@ func (handler *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
conn.SetReadLimit(int64(handler.maxMsgLen))
handler.messageType = websocket.TextMessage
handler.wg.Add(1)
defer handler.wg.Done()
@@ -64,7 +71,7 @@ func (handler *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
handler.conns[conn] = struct{}{}
handler.mutexConns.Unlock()
wsConn := newWSConn(conn, handler.pendingWriteNum, handler.maxMsgLen)
wsConn := newWSConn(conn, handler.pendingWriteNum, handler.maxMsgLen,handler.messageType)
agent := handler.newAgent(wsConn)
agent.Run()
@@ -76,6 +83,13 @@ func (handler *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
agent.OnClose()
}
func (server *WSServer) SetMessageType(messageType int){
server.messageType = messageType
if server.handler!= nil {
server.handler.SetMessageType(messageType)
}
}
func (server *WSServer) Start() {
ln, err := net.Listen("tcp", server.Addr)
if err != nil {

View File

@@ -9,6 +9,7 @@ import (
"github.com/duanhf2012/origin/profiler"
"github.com/duanhf2012/origin/service"
"github.com/duanhf2012/origin/util/timer"
"github.com/duanhf2012/origin/util/buildtime"
"io/ioutil"
slog "log"
"net/http"
@@ -38,6 +39,7 @@ func init() {
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM,syscall.Signal(10))
console.RegisterCommandBool("help",false,"<-help> This help.",usage)
console.RegisterCommandString("name","","<-name nodeName> Node's name.",setName)
console.RegisterCommandString("start","","<-start nodeid=nodeid> Run originserver.",startNode)
console.RegisterCommandString("stop","","<-stop nodeid=nodeid> Stop originserver process.",stopNode)
console.RegisterCommandString("config","","<-config path> Configuration file path.",setConfigPath)
@@ -53,13 +55,20 @@ func usage(val interface{}) error{
return nil
}
fmt.Fprintf(os.Stderr, `orgin version: orgin/2.14.20201029
Usage: originserver [-help] [-start node=1] [-stop] [-config path] [-pprof 0.0.0.0:6060]...
`)
if len(buildtime.GetBuildDateTime())>0 {
fmt.Fprintf(os.Stderr, "Welcome to Origin(build info: %s)\nUsage: originserver [-help] [-start node=1] [-stop] [-config path] [-pprof 0.0.0.0:6060]...\n",buildtime.GetBuildDateTime())
}else{
fmt.Fprintf(os.Stderr, "Welcome to Origin\nUsage: originserver [-help] [-start node=1] [-stop] [-config path] [-pprof 0.0.0.0:6060]...\n")
}
console.PrintDefaults()
return nil
}
func setName(val interface{}) error {
return nil
}
func setPprof(val interface{}) error {
listenAddr := val.(string)
if listenAddr==""{

View File

@@ -2,31 +2,33 @@ package service
import (
"fmt"
"reflect"
"sync/atomic"
"time"
"github.com/duanhf2012/origin/event"
"github.com/duanhf2012/origin/log"
rpcHandle "github.com/duanhf2012/origin/rpc"
"github.com/duanhf2012/origin/util/timer"
"reflect"
"sync/atomic"
"time"
)
const InitModuleId = 1e9
type IModule interface {
SetModuleId(moduleId uint32) bool
GetModuleId() uint32
AddModule(module IModule) (uint32,error)
AddModule(module IModule) (uint32, error)
GetModule(moduleId uint32) IModule
GetAncestor()IModule
GetAncestor() IModule
ReleaseModule(moduleId uint32)
NewModuleId() uint32
GetParent()IModule
GetParent() IModule
OnInit() error
OnRelease()
getBaseModule() IModule
GetService() IService
GetModuleName() string
GetEventProcessor()event.IEventProcessor
GetEventProcessor() event.IEventProcessor
NotifyEvent(ev event.IEvent)
}
@@ -38,25 +40,25 @@ type IModuleTimer interface {
type Module struct {
rpcHandle.IRpcHandler
moduleId uint32 //模块Id
moduleName string //模块名称
parent IModule //父亲
self IModule //自己
child map[uint32]IModule //孩子们
mapActiveTimer map[timer.ITimer]struct{}
moduleId uint32 //模块Id
moduleName string //模块名称
parent IModule //父亲
self IModule //自己
child map[uint32]IModule //孩子们
mapActiveTimer map[timer.ITimer]struct{}
mapActiveIdTimer map[uint64]timer.ITimer
dispatcher *timer.Dispatcher //timer
dispatcher *timer.Dispatcher //timer
//根结点
ancestor IModule //始祖
seedModuleId uint32 //模块id种子
descendants map[uint32]IModule //始祖的后裔们
ancestor IModule //始祖
seedModuleId uint32 //模块id种子
descendants map[uint32]IModule //始祖的后裔们
//事件管道
eventHandler event.IEventHandler
}
func (m *Module) SetModuleId(moduleId uint32) bool{
func (m *Module) SetModuleId(moduleId uint32) bool {
if m.moduleId > 0 {
return false
}
@@ -65,35 +67,35 @@ func (m *Module) SetModuleId(moduleId uint32) bool{
return true
}
func (m *Module) GetModuleId() uint32{
func (m *Module) GetModuleId() uint32 {
return m.moduleId
}
func (m *Module) GetModuleName() string{
func (m *Module) GetModuleName() string {
return m.moduleName
}
func (m *Module) OnInit() error{
return nil
func (m *Module) OnInit() error {
return nil
}
func (m *Module) AddModule(module IModule) (uint32,error){
func (m *Module) AddModule(module IModule) (uint32, error) {
//没有事件处理器不允许加入其他模块
if m.GetEventProcessor() == nil {
return 0,fmt.Errorf("module %+v Event Processor is nil", m.self)
return 0, fmt.Errorf("module %+v Event Processor is nil", m.self)
}
pAddModule := module.getBaseModule().(*Module)
if pAddModule.GetModuleId()==0 {
if pAddModule.GetModuleId() == 0 {
pAddModule.moduleId = m.NewModuleId()
}
if m.child == nil {
m.child = map[uint32]IModule{}
}
_,ok := m.child[module.GetModuleId()]
_, ok := m.child[module.GetModuleId()]
if ok == true {
return 0,fmt.Errorf("exists module id %d",module.GetModuleId())
return 0, fmt.Errorf("exists module id %d", module.GetModuleId())
}
pAddModule.IRpcHandler = m.IRpcHandler
pAddModule.self = module
@@ -105,17 +107,17 @@ func (m *Module) AddModule(module IModule) (uint32,error){
pAddModule.eventHandler.Init(m.eventHandler.GetEventProcessor())
err := module.OnInit()
if err != nil {
return 0,err
return 0, err
}
m.child[module.GetModuleId()] = module
m.ancestor.getBaseModule().(*Module).descendants[module.GetModuleId()] = module
log.SDebug("Add module ",module.GetModuleName()," completed")
return module.GetModuleId(),nil
log.SDebug("Add module ", module.GetModuleName(), " completed")
return module.GetModuleId(), nil
}
func (m *Module) ReleaseModule(moduleId uint32){
func (m *Module) ReleaseModule(moduleId uint32) {
pModule := m.GetModule(moduleId).getBaseModule().(*Module)
//释放子孙
@@ -123,19 +125,19 @@ func (m *Module) ReleaseModule(moduleId uint32){
m.ReleaseModule(id)
}
pModule.GetEventHandler().Destroy()
pModule.self.OnRelease()
pModule.GetEventHandler().Destroy()
log.SDebug("Release module ", pModule.GetModuleName())
for pTimer := range pModule.mapActiveTimer {
pTimer.Cancel()
}
for _,t := range pModule.mapActiveIdTimer {
for _, t := range pModule.mapActiveIdTimer {
t.Cancel()
}
delete(m.child,moduleId)
delete (m.ancestor.getBaseModule().(*Module).descendants,moduleId)
delete(m.child, moduleId)
delete(m.ancestor.getBaseModule().(*Module).descendants, moduleId)
//清理被删除的Module
pModule.self = nil
@@ -149,16 +151,17 @@ func (m *Module) ReleaseModule(moduleId uint32){
pModule.mapActiveIdTimer = nil
}
func (m *Module) NewModuleId() uint32{
m.ancestor.getBaseModule().(*Module).seedModuleId+=1
func (m *Module) NewModuleId() uint32 {
m.ancestor.getBaseModule().(*Module).seedModuleId += 1
return m.ancestor.getBaseModule().(*Module).seedModuleId
}
var timerSeedId uint32
func (m *Module) GenTimerId() uint64{
for{
newTimerId := (uint64(m.GetModuleId())<<32)|uint64(atomic.AddUint32(&timerSeedId,1))
if _,ok := m.mapActiveIdTimer[newTimerId];ok == true {
func (m *Module) GenTimerId() uint64 {
for {
newTimerId := (uint64(m.GetModuleId()) << 32) | uint64(atomic.AddUint32(&timerSeedId, 1))
if _, ok := m.mapActiveIdTimer[newTimerId]; ok == true {
continue
}
@@ -166,33 +169,32 @@ func (m *Module) GenTimerId() uint64{
}
}
func (m *Module) GetAncestor()IModule{
func (m *Module) GetAncestor() IModule {
return m.ancestor
}
func (m *Module) GetModule(moduleId uint32) IModule{
iModule,ok := m.GetAncestor().getBaseModule().(*Module).descendants[moduleId]
func (m *Module) GetModule(moduleId uint32) IModule {
iModule, ok := m.GetAncestor().getBaseModule().(*Module).descendants[moduleId]
if ok == false {
return nil
}
return iModule
}
func (m *Module) getBaseModule() IModule{
func (m *Module) getBaseModule() IModule {
return m
}
func (m *Module) GetParent()IModule{
func (m *Module) GetParent() IModule {
return m.parent
}
func (m *Module) OnCloseTimer(t timer.ITimer){
delete(m.mapActiveIdTimer,t.GetId())
delete(m.mapActiveTimer,t)
func (m *Module) OnCloseTimer(t timer.ITimer) {
delete(m.mapActiveIdTimer, t.GetId())
delete(m.mapActiveTimer, t)
}
func (m *Module) OnAddTimer(t timer.ITimer){
func (m *Module) OnAddTimer(t timer.ITimer) {
if t != nil {
if m.mapActiveTimer == nil {
m.mapActiveTimer = map[timer.ITimer]struct{}{}
@@ -204,33 +206,33 @@ func (m *Module) OnAddTimer(t timer.ITimer){
func (m *Module) AfterFunc(d time.Duration, cb func(*timer.Timer)) *timer.Timer {
if m.mapActiveTimer == nil {
m.mapActiveTimer =map[timer.ITimer]struct{}{}
m.mapActiveTimer = map[timer.ITimer]struct{}{}
}
return m.dispatcher.AfterFunc(d,nil,cb,m.OnCloseTimer,m.OnAddTimer)
return m.dispatcher.AfterFunc(d, nil, cb, m.OnCloseTimer, m.OnAddTimer)
}
func (m *Module) CronFunc(cronExpr *timer.CronExpr, cb func(*timer.Cron)) *timer.Cron {
if m.mapActiveTimer == nil {
m.mapActiveTimer =map[timer.ITimer]struct{}{}
m.mapActiveTimer = map[timer.ITimer]struct{}{}
}
return m.dispatcher.CronFunc(cronExpr,nil,cb,m.OnCloseTimer,m.OnAddTimer)
return m.dispatcher.CronFunc(cronExpr, nil, cb, m.OnCloseTimer, m.OnAddTimer)
}
func (m *Module) NewTicker(d time.Duration, cb func(*timer.Ticker)) *timer.Ticker {
if m.mapActiveTimer == nil {
m.mapActiveTimer =map[timer.ITimer]struct{}{}
m.mapActiveTimer = map[timer.ITimer]struct{}{}
}
return m.dispatcher.TickerFunc(d,nil,cb,m.OnCloseTimer,m.OnAddTimer)
return m.dispatcher.TickerFunc(d, nil, cb, m.OnCloseTimer, m.OnAddTimer)
}
func (m *Module) cb(*timer.Timer){
func (m *Module) cb(*timer.Timer) {
}
func (m *Module) SafeAfterFunc(timerId *uint64,d time.Duration, AdditionData interface{},cb func(uint64,interface{})) {
func (m *Module) SafeAfterFunc(timerId *uint64, d time.Duration, AdditionData interface{}, cb func(uint64, interface{})) {
if m.mapActiveIdTimer == nil {
m.mapActiveIdTimer = map[uint64]timer.ITimer{}
}
@@ -240,45 +242,45 @@ func (m *Module) SafeAfterFunc(timerId *uint64,d time.Duration, AdditionData int
}
*timerId = m.GenTimerId()
t := m.dispatcher.AfterFunc(d,cb,nil,m.OnCloseTimer,m.OnAddTimer)
t := m.dispatcher.AfterFunc(d, cb, nil, m.OnCloseTimer, m.OnAddTimer)
t.AdditionData = AdditionData
t.Id = *timerId
m.mapActiveIdTimer[*timerId] = t
}
func (m *Module) SafeCronFunc(cronId *uint64,cronExpr *timer.CronExpr, AdditionData interface{}, cb func(uint64,interface{})) {
func (m *Module) SafeCronFunc(cronId *uint64, cronExpr *timer.CronExpr, AdditionData interface{}, cb func(uint64, interface{})) {
if m.mapActiveIdTimer == nil {
m.mapActiveIdTimer = map[uint64]timer.ITimer{}
}
*cronId = m.GenTimerId()
c := m.dispatcher.CronFunc(cronExpr,cb,nil,m.OnCloseTimer,m.OnAddTimer)
c := m.dispatcher.CronFunc(cronExpr, cb, nil, m.OnCloseTimer, m.OnAddTimer)
c.AdditionData = AdditionData
c.Id = *cronId
m.mapActiveIdTimer[*cronId] = c
}
func (m *Module) SafeNewTicker(tickerId *uint64,d time.Duration, AdditionData interface{}, cb func(uint64,interface{})) {
func (m *Module) SafeNewTicker(tickerId *uint64, d time.Duration, AdditionData interface{}, cb func(uint64, interface{})) {
if m.mapActiveIdTimer == nil {
m.mapActiveIdTimer = map[uint64]timer.ITimer{}
}
*tickerId = m.GenTimerId()
t := m.dispatcher.TickerFunc(d,cb,nil,m.OnCloseTimer,m.OnAddTimer)
t := m.dispatcher.TickerFunc(d, cb, nil, m.OnCloseTimer, m.OnAddTimer)
t.AdditionData = AdditionData
t.Id = *tickerId
m.mapActiveIdTimer[*tickerId] = t
}
func (m *Module) CancelTimerId(timerId *uint64) bool{
func (m *Module) CancelTimerId(timerId *uint64) bool {
if m.mapActiveIdTimer == nil {
log.SError("mapActiveIdTimer is nil")
return false
}
t,ok := m.mapActiveIdTimer[*timerId]
t, ok := m.mapActiveIdTimer[*timerId]
if ok == false {
log.SError("cannot find timer id ",timerId)
log.SError("cannot find timer id ", timerId)
return false
}
@@ -287,23 +289,21 @@ func (m *Module) CancelTimerId(timerId *uint64) bool{
return true
}
func (m *Module) OnRelease(){
func (m *Module) OnRelease() {
}
func (m *Module) GetService() IService {
return m.GetAncestor().(IService)
}
func (m *Module) GetEventProcessor() event.IEventProcessor{
func (m *Module) GetEventProcessor() event.IEventProcessor {
return m.eventHandler.GetEventProcessor()
}
func (m *Module) NotifyEvent(ev event.IEvent){
func (m *Module) NotifyEvent(ev event.IEvent) {
m.eventHandler.NotifyEvent(ev)
}
func (m *Module) GetEventHandler() event.IEventHandler{
func (m *Module) GetEventHandler() event.IEventHandler {
return m.eventHandler
}
}

View File

@@ -65,23 +65,25 @@ func (s *Session) NextSeq(db string, collection string, id interface{}) (int, er
ctxTimeout, cancel := s.GetDefaultContext()
defer cancel()
err := s.Client.Database(db).Collection(collection).FindOneAndUpdate(ctxTimeout, bson.M{"_id": id}, bson.M{"$inc": bson.M{"Seq": 1}}).Decode(&res)
after := options.After
updateOpts := options.FindOneAndUpdateOptions{ReturnDocument: &after}
err := s.Client.Database(db).Collection(collection).FindOneAndUpdate(ctxTimeout, bson.M{"_id": id}, bson.M{"$inc": bson.M{"Seq": 1}},&updateOpts).Decode(&res)
return res.Seq, err
}
//indexKeys[索引][每个索引key字段]
func (s *Session) EnsureIndex(db string, collection string, indexKeys [][]string, bBackground bool) error {
return s.ensureIndex(db, collection, indexKeys, bBackground, false)
func (s *Session) EnsureIndex(db string, collection string, indexKeys [][]string, bBackground bool,sparse bool) error {
return s.ensureIndex(db, collection, indexKeys, bBackground, false,sparse)
}
//indexKeys[索引][每个索引key字段]
func (s *Session) EnsureUniqueIndex(db string, collection string, indexKeys [][]string, bBackground bool) error {
return s.ensureIndex(db, collection, indexKeys, bBackground, true)
func (s *Session) EnsureUniqueIndex(db string, collection string, indexKeys [][]string, bBackground bool,sparse bool) error {
return s.ensureIndex(db, collection, indexKeys, bBackground, true,sparse)
}
//keys[索引][每个索引key字段]
func (s *Session) ensureIndex(db string, collection string, indexKeys [][]string, bBackground bool, unique bool) error {
func (s *Session) ensureIndex(db string, collection string, indexKeys [][]string, bBackground bool, unique bool,sparse bool) error {
var indexes []mongo.IndexModel
for _, keys := range indexKeys {
keysDoc := bsonx.Doc{}
@@ -89,10 +91,11 @@ func (s *Session) ensureIndex(db string, collection string, indexKeys [][]string
keysDoc = keysDoc.Append(key, bsonx.Int32(1))
}
indexes = append(indexes, mongo.IndexModel{
Keys: keysDoc,
Options: options.Index().SetUnique(unique).SetBackground(bBackground),
})
options:= options.Index().SetUnique(unique).SetBackground(bBackground)
if sparse == true {
options.SetSparse(true)
}
indexes = append(indexes, mongo.IndexModel{Keys: keysDoc, Options:options })
}
ctxTimeout, cancel := context.WithTimeout(context.Background(), s.maxOperatorTimeOut)

View File

@@ -175,10 +175,12 @@ func (slf *HttpSession) Write(msg []byte) {
func (slf *HttpSession) WriteJsonDone(statusCode int,msgJson interface{}) error {
msg, err := json.Marshal(msgJson)
if err == nil {
slf.Write(msg)
if err != nil {
return err
}
slf.statusCode = statusCode
slf.Write(msg)
slf.Done()
return err
}

View File

@@ -8,6 +8,7 @@ import (
"github.com/duanhf2012/origin/network/processor"
"github.com/duanhf2012/origin/node"
"github.com/duanhf2012/origin/service"
"sync/atomic"
"sync"
"time"
"runtime"
@@ -42,12 +43,12 @@ const Default_ReadDeadline = 180 //30s
const Default_WriteDeadline = 180 //30s
const (
MaxNodeId = 1<<10 - 1 //Uint10
MaxSeed = 1<<22 - 1 //MaxUint24
MaxNodeId = 1<<14 - 1 //最大值 16383
MaxSeed = 1<<19 - 1 //最大值 524287
MaxTime = 1<<31 - 1 //最大值 2147483647
)
var seed uint32
var seedLocker sync.Mutex
type TcpPack struct {
Type TcpPackType //0表示连接 1表示断开 2表示数据
@@ -66,16 +67,14 @@ func (tcpService *TcpService) genId() uint64 {
panic("nodeId exceeds the maximum!")
}
seedLocker.Lock()
seed = (seed+1)%MaxSeed
seedLocker.Unlock()
nowTime := uint64(time.Now().Second())
return (uint64(node.GetNodeId())<<54)|(nowTime<<22)|uint64(seed)
newSeed := atomic.AddUint32(&seed,1) % MaxSeed
nowTime := uint64(time.Now().Unix())%MaxTime
return (uint64(node.GetNodeId())<<50)|(nowTime<<19)|uint64(newSeed)
}
func GetNodeId(agentId uint64) int {
return int(agentId>>54)
return int(agentId>>50)
}
func (tcpService *TcpService) OnInit() error{

View File

@@ -7,19 +7,27 @@ import (
"github.com/duanhf2012/origin/network"
"github.com/duanhf2012/origin/network/processor"
"github.com/duanhf2012/origin/service"
"github.com/duanhf2012/origin/node"
"sync"
"sync/atomic"
"time"
)
type WSService struct {
service.Service
wsServer network.WSServer
mapClientLocker sync.RWMutex
mapClient map[uint64] *WSClient
initClientId uint64
process processor.IProcessor
}
var seed uint32
type WSPackType int8
const(
WPT_Connected WSPackType = 0
@@ -32,6 +40,12 @@ const Default_WS_MaxConnNum = 3000
const Default_WS_PendingWriteNum = 10000
const Default_WS_MaxMsgLen = 65535
const (
MaxNodeId = 1<<14 - 1 //最大值 16383
MaxSeed = 1<<19 - 1 //最大值 524287
MaxTime = 1<<31 - 1 //最大值 2147483647
)
type WSClient struct {
id uint64
wsConn *network.WSConn
@@ -46,6 +60,7 @@ type WSPack struct {
}
func (ws *WSService) OnInit() error{
iConfig := ws.GetServiceCfg()
if iConfig == nil {
return fmt.Errorf("%s service config is error!", ws.GetName())
@@ -80,6 +95,10 @@ func (ws *WSService) OnInit() error{
return nil
}
func (ws *WSService) SetMessageType(messageType int){
ws.wsServer.SetMessageType(messageType)
}
func (ws *WSService) WSEventHandler(ev event.IEvent) {
pack := ev.(*event.Event).Data.(*WSPack)
switch pack.Type {
@@ -88,9 +107,9 @@ func (ws *WSService) WSEventHandler(ev event.IEvent) {
case WPT_DisConnected:
pack.MsgProcessor.DisConnectedRoute(pack.ClientId)
case WPT_UnknownPack:
pack.MsgProcessor.UnknownMsgRoute(pack.Data,pack.ClientId)
pack.MsgProcessor.UnknownMsgRoute(pack.ClientId,pack.Data)
case WPT_Pack:
pack.MsgProcessor.MsgRoute(pack.Data, pack.ClientId)
pack.MsgProcessor.MsgRoute(pack.ClientId,pack.Data)
}
}
@@ -99,20 +118,30 @@ func (ws *WSService) SetProcessor(process processor.IProcessor,handler event.IEv
ws.RegEventReceiverFunc(event.Sys_Event_WebSocket,handler, ws.WSEventHandler)
}
func (ws *WSService) genId() uint64 {
if node.GetNodeId()>MaxNodeId{
panic("nodeId exceeds the maximum!")
}
newSeed := atomic.AddUint32(&seed,1) % MaxSeed
nowTime := uint64(time.Now().Unix())%MaxTime
return (uint64(node.GetNodeId())<<50)|(nowTime<<19)|uint64(newSeed)
}
func (ws *WSService) NewWSClient(conn *network.WSConn) network.Agent {
ws.mapClientLocker.Lock()
defer ws.mapClientLocker.Unlock()
for {
ws.initClientId+=1
_,ok := ws.mapClient[ws.initClientId]
clientId := ws.genId()
_,ok := ws.mapClient[clientId]
if ok == true {
continue
}
pClient := &WSClient{wsConn:conn, id: ws.initClientId}
pClient := &WSClient{wsConn:conn, id: clientId}
pClient.wsService = ws
ws.mapClient[ws.initClientId] = pClient
ws.mapClient[clientId] = pClient
return pClient
}
@@ -131,7 +160,7 @@ func (slf *WSClient) Run() {
log.Debug("read client id %d is error:%+v",slf.id,err)
break
}
data,err:=slf.wsService.process.Unmarshal(bytes)
data,err:=slf.wsService.process.Unmarshal(slf.id,bytes)
if err != nil {
slf.wsService.NotifyEvent(&event.Event{Type:event.Sys_Event_WebSocket,Data:&WSPack{ClientId:slf.id,Type:WPT_UnknownPack,Data:bytes,MsgProcessor:slf.wsService.process}})
continue
@@ -156,7 +185,7 @@ func (ws *WSService) SendMsg(clientid uint64,msg interface{}) error{
}
ws.mapClientLocker.Unlock()
bytes,err := ws.process.Marshal(msg)
bytes,err := ws.process.Marshal(clientid,msg)
if err != nil {
return err
}

15
util/buildtime/build.go Normal file
View File

@@ -0,0 +1,15 @@
package buildtime
/*
//查询buildtime包中的位置在github.com/duanhf2012/origin/util/buildtime.BuildTime中
go tool nm ./originserver.exe |grep buildtime
//编译传入编译时间信息
go build -ldflags "-X 'github.com/duanhf2012/origin/util/buildtime.BuildTime=20200101'"
*/
var BuildTime string
func GetBuildDateTime() string {
return BuildTime
}