zhangdeman/websocket
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

升级server逻辑, 多模块监听同一个端口,除了共享GIN实例外,其他均按照模块进行隔离

Browse Source
This commit is contained in:
白茶清欢 2021-04-18 00:43:19 +08:00
parent 63be630c68
commit 4177c01939
1 changed files with 139 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
construct.go
View File

@ -9,7 +9,6 @@ package websocket
import ( import (
"fmt" "fmt"
"sync"
"github.com/go-developer/gopkg/logger" "github.com/go-developer/gopkg/logger"
@ -25,8 +24,6 @@ import (
"gopkg.in/olahol/melody.v1" "gopkg.in/olahol/melody.v1"
"github.com/go-developer/gopkg/easylock"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"github.com/go-developer/websocket/abstract" "github.com/go-developer/websocket/abstract"
"github.com/pkg/errors" "github.com/pkg/errors"
@ -46,7 +43,9 @@ type Server struct {
var ( var (
// ginRouterTable 表 // ginRouterTable 表
ginRouterTable map[int]*Server ginRouterTable map[int]*gin.Engine
// wsServerTable 表
wsServerTable map[int]map[string]*Server
// commandTable 指令表 // commandTable 指令表
commandTable map[string]map[string]abstract.ICommand commandTable map[string]map[string]abstract.ICommand
// 服务启停的信号 // 服务启停的信号
@ -62,17 +61,29 @@ func NewWebsocketServe(wsInstanceList ...abstract.IWebsocket) error {
if len(wsInstanceList) == 0 { if len(wsInstanceList) == 0 {
return errors.WithStack(errors.New("register websocket server list is empty")) return errors.WithStack(errors.New("register websocket server list is empty"))
} }
lock := easylock.NewLock() ginRouterTable = make(map[int]*gin.Engine)
ginRouterTable = make(map[int]*Server) wsServerTable = make(map[int]map[string]*Server)
commandTable = make(map[string]map[string]abstract.ICommand) commandTable = make(map[string]map[string]abstract.ICommand)
wg := &sync.WaitGroup{}
wg.Add(len(wsInstanceList))
for _, wsInstance := range wsInstanceList { for _, wsInstance := range wsInstanceList {
go func(wsInstance abstract.IWebsocket) { initServer(wsInstance)
defer wg.Done() }
run()
return nil
}
// 初始化server
func initServer(wsInstance abstract.IWebsocket) {
// 初始化ws server // 初始化ws server
_ = lock.Lock() // 初始化 gin 路由表
if _, exist := ginRouterTable[wsInstance.GetServerPort()]; !exist { if _, exist := ginRouterTable[wsInstance.GetServerPort()]; !exist {
ginRouterTable[wsInstance.GetServerPort()] = gin.Default()
}
// 初始化WS-Server表
if _, exist := wsServerTable[wsInstance.GetServerPort()]; !exist {
wsServerTable[wsInstance.GetServerPort()] = make(map[string]*Server)
}
if _, exist := wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()]; !exist {
// 生成并存储 WS-Server
wsSetConfigList := wsInstance.GetWSServerConfig() wsSetConfigList := wsInstance.GetWSServerConfig()
if nil == wsSetConfigList { if nil == wsSetConfigList {
wsSetConfigList = make([]config.SetWSServerConfig, 0) wsSetConfigList = make([]config.SetWSServerConfig, 0)
@ -115,69 +126,70 @@ func NewWebsocketServe(wsInstanceList ...abstract.IWebsocket) error {
panic(wsInstance.GetModuleFlag() + " 模块开启了控制台日志记录,日志初始化失败, 失败原因 : " + err.Error()) panic(wsInstance.GetModuleFlag() + " 模块开启了控制台日志记录,日志初始化失败, 失败原因 : " + err.Error())
} }
} }
ginRouterTable[wsInstance.GetServerPort()] = s wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()] = s
} }
// 初始化指令存储表
if _, exist := commandTable[wsInstance.GetModuleFlag()]; !exist { if _, exist := commandTable[wsInstance.GetModuleFlag()]; !exist {
commandTable[wsInstance.GetModuleFlag()] = make(map[string]abstract.ICommand) commandTable[wsInstance.GetModuleFlag()] = make(map[string]abstract.ICommand)
} }
_ = lock.Unlock() routerGroup := ginRouterTable[wsInstance.GetServerPort()].Group(wsInstance.GetModuleFlag())
routerGroup := ginRouterTable[wsInstance.GetServerPort()].ginRouter.Group(wsInstance.GetModuleFlag())
// 注册路由 // 注册路由
for _, path := range wsInstance.HandshakeURL() { for _, path := range wsInstance.HandshakeURL() {
routerGroup.GET(path, func(ctx *gin.Context) { routerGroup.GET(path, func(ctx *gin.Context) {
parameter := map[string]interface{}{ parameter := map[string]interface{}{
"ws_context": context.NewContext(ctx, wsInstance.GetModuleFlag(), nil), "ws_context": context.NewContext(ctx, wsInstance.GetModuleFlag(), nil),
} }
_ = ginRouterTable[wsInstance.GetServerPort()].wsServer.HandleRequestWithKeys(ctx.Writer, ctx.Request, parameter) _ = wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].wsServer.HandleRequestWithKeys(ctx.Writer, ctx.Request, parameter)
}) })
} }
// 注册回调函数 // 注册回调函数
// 1. 建立连接的函数 // 1. 建立连接的函数
ginRouterTable[wsInstance.GetServerPort()].wsServer.HandleConnect(func(session *melody.Session) { wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].wsServer.HandleConnect(func(session *melody.Session) {
ctxInterface, _ := session.Get("ws_context") ctxInterface, _ := session.Get("ws_context")
ctx := ctxInterface.(*context.WSContext) ctx := ctxInterface.(*context.WSContext)
ctx.Session = session ctx.Session = session
if err := wsInstance.Connect(ctx); nil == err { if err := wsInstance.Connect(ctx); nil == err {
if ginRouterTable[wsInstance.GetServerPort()].conf.StoreConnection && nil != ginRouterTable[wsInstance.GetServerPort()].conf.ConnectionManager { if wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.StoreConnection && nil != wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.ConnectionManager {
ginRouterTable[wsInstance.GetServerPort()].conf.ConnectionManager.Store(ctx) wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.ConnectionManager.Store(ctx)
} }
} }
}) })
// 2. 指令处理的函数 // 2. 指令处理的函数
ginRouterTable[wsInstance.GetServerPort()].wsServer.HandleMessage(func(session *melody.Session, bytes []byte) { wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].wsServer.HandleMessage(func(session *melody.Session, bytes []byte) {
// TODO : 增加指令回调失败的callback // TODO : 增加指令回调失败的callback
ctxInterface, _ := session.Get("ws_context") ctxInterface, _ := session.Get("ws_context")
ctx := ctxInterface.(*context.WSContext) ctx := ctxInterface.(*context.WSContext)
_ = dispatchCommand(ctx, bytes) _ = dispatchCommand(ctx, bytes)
}) })
// 3, 关闭连接的处理函数 // 3, 关闭连接的处理函数
ginRouterTable[wsInstance.GetServerPort()].wsServer.HandleClose(func(session *melody.Session, i int, s string) error { wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].wsServer.HandleClose(func(session *melody.Session, i int, s string) error {
ctxInterface, _ := session.Get("ws_context") ctxInterface, _ := session.Get("ws_context")
ctx := ctxInterface.(*context.WSContext) ctx := ctxInterface.(*context.WSContext)
defer func() { defer func() {
if !ginRouterTable[wsInstance.GetServerPort()].conf.StoreConnection || nil == ginRouterTable[wsInstance.GetServerPort()].conf.ConnectionManager { if !wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.StoreConnection || nil == wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.ConnectionManager {
return return
} }
ginRouterTable[wsInstance.GetServerPort()].conf.ConnectionManager.Del(ctx) wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.ConnectionManager.Del(ctx, "")
}() }()
return wsInstance.Close(ctx, i, s) return wsInstance.Close(ctx, i, s)
}) })
// 4. 断开连接的处理函数 // 4. 断开连接的处理函数
ginRouterTable[wsInstance.GetServerPort()].wsServer.HandleDisconnect(func(session *melody.Session) { wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].wsServer.HandleDisconnect(func(session *melody.Session) {
ctxInterface, _ := session.Get("ws_context") ctxInterface, _ := session.Get("ws_context")
ctx := ctxInterface.(*context.WSContext) ctx := ctxInterface.(*context.WSContext)
defer func() { defer func() {
if nil == ginRouterTable[wsInstance.GetServerPort()].conf.ConnectionManager { if nil == wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.ConnectionManager {
return return
} }
ginRouterTable[wsInstance.GetServerPort()].conf.ConnectionManager.Del(ctx) wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].conf.ConnectionManager.Del(ctx, "")
}() }()
wsInstance.Disconnect(ctx) wsInstance.Disconnect(ctx)
}) })
// 注册指令 // 注册指令
for _, cmd := range wsInstance.GetCommandList() { for _, cmd := range wsInstance.GetCommandList() {
if nil != ginRouterTable[wsInstance.GetServerPort()].loggerInstance { if nil != wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].loggerInstance {
ginRouterTable[wsInstance.GetServerPort()].loggerInstance.Debug( wsServerTable[wsInstance.GetServerPort()][wsInstance.GetModuleFlag()].loggerInstance.Debug(
"长连接指令注册成功", "长连接指令注册成功",
zap.String("module", wsInstance.GetModuleFlag()), zap.String("module", wsInstance.GetModuleFlag()),
zap.String("command", cmd.GetCommand()), zap.String("command", cmd.GetCommand()),
@ -186,15 +198,10 @@ func NewWebsocketServe(wsInstanceList ...abstract.IWebsocket) error {
commandTable[wsInstance.GetModuleFlag()][cmd.GetCommand()] = cmd commandTable[wsInstance.GetModuleFlag()][cmd.GetCommand()] = cmd
} }
go func() { go func() {
if err := ginRouterTable[wsInstance.GetServerPort()].ginRouter.Run(fmt.Sprintf(":%d", wsInstance.GetServerPort())); nil != err { if err := ginRouterTable[wsInstance.GetServerPort()].Run(fmt.Sprintf(":%d", wsInstance.GetServerPort())); nil != err {
panic(err) panic(err)
} }
}() }()
}(wsInstance)
}
wg.Wait()
run()
return nil
} }
// dispatchCommand 调度command ... // dispatchCommand 调度command ...