mirror of
https://github.com/duanhf2012/origin.git
synced 2026-02-05 23:54:53 +08:00
Compare commits
16 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
be3daf19f9 | ||
|
|
aa91c7bf1b | ||
|
|
7fe73e55fb | ||
|
|
e5ceaa9e76 | ||
|
|
97c55ada71 | ||
|
|
776b234022 | ||
|
|
a4f425bd69 | ||
|
|
ee54862be2 | ||
|
|
13642d7402 | ||
|
|
18d620118e | ||
|
|
f02592d126 | ||
|
|
0f890887f7 | ||
|
|
c5c05b6ae9 | ||
|
|
68b891df51 | ||
|
|
63199bf862 | ||
|
|
4d5d45d555 |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -10,3 +10,4 @@
|
||||
|
||||
# Output of the go coverage tool, specifically when used with LiteIDE
|
||||
*.out
|
||||
.DS_Store
|
||||
|
||||
@@ -87,6 +87,9 @@ service.json如下:
|
||||
---------------
|
||||
```
|
||||
{
|
||||
"Global": {
|
||||
"AreaId": 1
|
||||
},
|
||||
"Service":{
|
||||
"HttpService":{
|
||||
"ListenAddr":"0.0.0.0:9402",
|
||||
@@ -157,7 +160,7 @@ service.json如下:
|
||||
```
|
||||
|
||||
---------------
|
||||
以上配置分为两个部分:Service与NodeService,NodeService中配置的对应结点中服务的配置,如果启动程序中根据nodeid查找该域的对应的服务,如果找不到时,从Service公共部分查找。
|
||||
以上配置分为两个部分:Global,Service与NodeService。Global是全局配置,在任何服务中都可以通过cluster.GetCluster().GetGlobalCfg()获取,NodeService中配置的对应结点中服务的配置,如果启动程序中根据nodeid查找该域的对应的服务,如果找不到时,从Service公共部分查找。
|
||||
|
||||
**HttpService配置**
|
||||
* ListenAddr:Http监听地址
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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){
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
15
node/node.go
15
node/node.go
@@ -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==""{
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -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
15
util/buildtime/build.go
Normal 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
|
||||
}
|
||||
Reference in New Issue
Block a user