diff --git a/go.mod b/go.mod index f704c6e..e75dfc5 100644 --- a/go.mod +++ b/go.mod @@ -6,12 +6,14 @@ require ( github.com/bytedance/sonic v1.10.2 github.com/dsnet/golib/unitconv v1.0.2 github.com/gin-gonic/gin v1.9.1 + github.com/gobwas/ws v1.3.2 github.com/golang/protobuf v1.5.3 github.com/google/uuid v1.4.0 github.com/gorilla/websocket v1.5.1 github.com/jhump/protoreflect v1.15.4 github.com/lxzan/gws v1.7.0 github.com/nats-io/nats.go v1.31.0 + github.com/panjf2000/gnet/v2 v2.3.4 github.com/patrickmn/go-cache v2.1.0+incompatible github.com/redis/go-redis/v9 v9.4.0 github.com/sirupsen/logrus v1.9.3 @@ -39,6 +41,8 @@ require ( github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.14.0 // indirect github.com/go-sql-driver/mysql v1.7.0 // indirect + github.com/gobwas/httphead v0.1.0 // indirect + github.com/gobwas/pool v0.2.1 // indirect github.com/goccy/go-json v0.10.2 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/hashicorp/hcl v1.0.0 // indirect @@ -65,6 +69,7 @@ require ( github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.11 // indirect + github.com/valyala/bytebufferpool v1.0.0 // indirect go.etcd.io/etcd/client/pkg/v3 v3.5.11 // indirect go.uber.org/atomic v1.9.0 // indirect go.uber.org/multierr v1.9.0 // indirect @@ -80,5 +85,6 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20231106174013-bbf56f31fb17 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20231120223509-83a465c0220f // indirect gopkg.in/ini.v1 v1.67.0 // indirect + gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 8ec8b99..000c8f1 100644 --- a/go.sum +++ b/go.sum @@ -46,6 +46,12 @@ github.com/go-playground/validator/v10 v10.14.0 h1:vgvQWe3XCz3gIeFDm/HnTIbj6UGmg github.com/go-playground/validator/v10 v10.14.0/go.mod h1:9iXMNT7sEkjXb0I+enO7QXmzG6QCsPWY4zveKFVRSyU= github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc= github.com/go-sql-driver/mysql v1.7.0/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI= +github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU= +github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM= +github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og= +github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= +github.com/gobwas/ws v1.3.2 h1:zlnbNHxumkRvfPWgfXu8RBwyNR1x8wh9cf5PTOCqs9Q= +github.com/gobwas/ws v1.3.2/go.mod h1:hRKAFb8wOxFROYNsT1bqfWnhX+b5MFeJM9r2ZSwg/KY= github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= @@ -106,6 +112,9 @@ github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI= github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/panjf2000/ants/v2 v2.8.2 h1:D1wfANttg8uXhC9149gRt1PDQ+dLVFjNXkCEycMcvQQ= +github.com/panjf2000/gnet/v2 v2.3.4 h1:+ASHt+Wxr0KIzlk5FsLBbegCc4US7iVCdZ1QbUyw17g= +github.com/panjf2000/gnet/v2 v2.3.4/go.mod h1:0mTLWq4zMEXyQ35BY094dNWYnXfIdDg0mOlmZJflaXE= github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc= github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ= github.com/pelletier/go-toml/v2 v2.1.0 h1:FnwAJ4oYMvbT/34k9zzHuZNrhlz48GB3/s6at6/MHO4= @@ -150,6 +159,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.11 h1:BMaWp1Bb6fHwEtbplGBGJ498wD+LKlNSl25MjdZY4dU= github.com/ugorji/go/codec v1.2.11/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= +github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= +github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= @@ -240,6 +251,8 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA= gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= +gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= +gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/gateway_ws/gnet_server/codec.go b/internal/gateway_ws/gnet_server/codec.go new file mode 100644 index 0000000..2379aaa --- /dev/null +++ b/internal/gateway_ws/gnet_server/codec.go @@ -0,0 +1,161 @@ +package gnet_server + +import ( + "bytes" + "fmt" + "github.com/gobwas/ws" + "github.com/gobwas/ws/wsutil" + "github.com/panjf2000/gnet/v2" + "github.com/panjf2000/gnet/v2/pkg/logging" + "io" +) + +type wsCodec struct { + upgraded bool // 链接是否升级 + buf bytes.Buffer // 从实际socket中读取到的数据缓存 + wsMsgBuf wsMessageBuf // ws 消息缓存 +} + +type wsMessageBuf struct { + firstHeader *ws.Header + curHeader *ws.Header + cachedBuf bytes.Buffer +} + +type readWrite struct { + io.Reader + io.Writer +} + +func (w *wsCodec) upgrade(c gnet.Conn) (ok bool, action gnet.Action) { + if w.upgraded { + ok = true + return + } + buf := &w.buf + tmpReader := bytes.NewReader(buf.Bytes()) + oldLen := tmpReader.Len() + logging.Infof("do Upgrade") + + hs, err := ws.Upgrade(readWrite{tmpReader, c}) + skipN := oldLen - tmpReader.Len() + if err != nil { + if err == io.EOF || err == io.ErrUnexpectedEOF { //数据不完整 + return + } + buf.Next(skipN) + logging.Infof("conn[%v] [err=%v]", c.RemoteAddr().String(), err.Error()) + action = gnet.Close + return + } + buf.Next(skipN) + logging.Infof("conn[%v] upgrade websocket protocol! Handshake: %v", c.RemoteAddr().String(), hs) + if err != nil { + logging.Infof("conn[%v] [err=%v]", c.RemoteAddr().String(), err.Error()) + action = gnet.Close + return + } + ok = true + w.upgraded = true + return +} +func (w *wsCodec) readBufferBytes(c gnet.Conn) gnet.Action { + size := c.InboundBuffered() + buf := make([]byte, size, size) + read, err := c.Read(buf) + if err != nil { + logging.Infof("read err! %w", err) + return gnet.Close + } + if read < size { + logging.Infof("read bytes len err! size: %d read: %d", size, read) + return gnet.Close + } + w.buf.Write(buf) + return gnet.None +} +func (w *wsCodec) Decode(c gnet.Conn) (outs []wsutil.Message, err error) { + fmt.Println("do Decode") + messages, err := w.readWsMessages() + if err != nil { + logging.Infof("Error reading message! %v", err) + return nil, err + } + if messages == nil || len(messages) <= 0 { //没有读到完整数据 不处理 + return + } + for _, message := range messages { + if message.OpCode.IsControl() { + err = wsutil.HandleClientControlMessage(c, message) + if err != nil { + return + } + continue + } + if message.OpCode == ws.OpText || message.OpCode == ws.OpBinary { + outs = append(outs, message) + } + } + return +} + +func (w *wsCodec) readWsMessages() (messages []wsutil.Message, err error) { + msgBuf := &w.wsMsgBuf + in := &w.buf + for { + if msgBuf.curHeader == nil { + if in.Len() < ws.MinHeaderSize { //头长度至少是2 + return + } + var head ws.Header + if in.Len() >= ws.MaxHeaderSize { + head, err = ws.ReadHeader(in) + if err != nil { + return messages, err + } + } else { //有可能不完整，构建新的 reader 读取 head 读取成功才实际对 in 进行读操作 + tmpReader := bytes.NewReader(in.Bytes()) + oldLen := tmpReader.Len() + head, err = ws.ReadHeader(tmpReader) + skipN := oldLen - tmpReader.Len() + if err != nil { + if err == io.EOF || err == io.ErrUnexpectedEOF { //数据不完整 + return messages, nil + } + in.Next(skipN) + return nil, err + } + in.Next(skipN) + } + + msgBuf.curHeader = &head + err = ws.WriteHeader(&msgBuf.cachedBuf, head) + if err != nil { + return nil, err + } + } + dataLen := (int)(msgBuf.curHeader.Length) + if dataLen > 0 { + if in.Len() >= dataLen { + _, err = io.CopyN(&msgBuf.cachedBuf, in, int64(dataLen)) + if err != nil { + return + } + } else { //数据不完整 + fmt.Println(in.Len(), dataLen) + logging.Infof("incomplete data") + return + } + } + if msgBuf.curHeader.Fin { //当前 header 已经是一个完整消息 + messages, err = wsutil.ReadClientMessage(&msgBuf.cachedBuf, messages) + if err != nil { + return nil, err + } + msgBuf.cachedBuf.Reset() + } else { + logging.Infof("The data is split into multiple frames") + } + msgBuf.curHeader = nil + } +} diff --git a/internal/gateway_ws/gnet_server/gnet_conn.go b/internal/gateway_ws/gnet_server/gnet_conn.go new file mode 100644 index 0000000..b54643d --- /dev/null +++ b/internal/gateway_ws/gnet_server/gnet_conn.go @@ -0,0 +1,23 @@ +package gnet_server + +import ( + "github.com/gobwas/ws" + "github.com/gobwas/ws/wsutil" + "github.com/panjf2000/gnet/v2" +) + +type gnetConn struct { + conn gnet.Conn + uid string + wsCodec *wsCodec +} + +func newGnetConn(conn gnet.Conn) *gnetConn { + return &gnetConn{ + conn: conn, + } +} + +func (c *gnetConn) Write(msg []byte) error { + return wsutil.WriteServerMessage(c.conn, ws.OpBinary, msg) +} diff --git a/internal/gateway_ws/gnet_server/websocket.go b/internal/gateway_ws/gnet_server/websocket.go new file mode 100644 index 0000000..927c907 --- /dev/null +++ b/internal/gateway_ws/gnet_server/websocket.go @@ -0,0 +1,184 @@ +package gnet_server + +import ( + "context" + "flag" + "fmt" + "github.com/gobwas/ws" + "log" + "sonet/pkg/protocol" + "sonet/pkg/protocol/session" + "sonet/pkg/utils/logger" + "sync/atomic" + "time" + + "github.com/panjf2000/gnet/v2" + "github.com/panjf2000/gnet/v2/pkg/logging" +) + +type GnetWsServer struct { + gnet.BuiltinEventEngine + + addr string + multicore bool + eng gnet.Engine + connected int64 + requests chan *protocol.Payload +} + +func NewGnetWsServer() *GnetWsServer { + return &GnetWsServer{ + requests: make(chan *protocol.Payload, 1024*16), + } +} + +func (wss *GnetWsServer) Init() { + +} + +// dispatch 多个 goroutine 进行请求 dispatch +func (wss *GnetWsServer) dispatch(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case payload := <-wss.requests: + fmt.Println(payload.Header) + } + } +} + +func (wss *GnetWsServer) OnBoot(eng gnet.Engine) gnet.Action { + wss.eng = eng + logging.Infof("echo server with multi-core=%t is listening on %s", wss.multicore, wss.addr) + return gnet.None +} + +func (wss *GnetWsServer) OnOpen(c gnet.Conn) ([]byte, gnet.Action) { + // wsCodec, uid, gnetConn + conn := newGnetConn(c) + conn.wsCodec = new(wsCodec) + c.SetContext(conn) + + // c.SetContext(new(wsCodec)) + atomic.AddInt64(&wss.connected, 1) + return nil, gnet.None +} + +func (wss *GnetWsServer) OnClose(c gnet.Conn, err error) (action gnet.Action) { + if err != nil { + logging.Warnf("error occurred on connection=%s, %v\n", c.RemoteAddr().String(), err) + } + atomic.AddInt64(&wss.connected, -1) + logging.Infof("conn[%v] disconnected", c.RemoteAddr().String()) + return gnet.None +} + +func (wss *GnetWsServer) OnTraffic(c gnet.Conn) (action gnet.Action) { + conn := c.Context().(*gnetConn) + if conn.wsCodec.readBufferBytes(c) == gnet.Close { + return gnet.Close + } + ok, action := conn.wsCodec.upgrade(c) + if !ok { + return + } + + if conn.wsCodec.buf.Len() <= 0 { + return gnet.None + } + messages, err := conn.wsCodec.Decode(c) + if err != nil { + return gnet.Close + } + if messages == nil { + return + } + authorized := conn.uid != "" + + for _, message := range messages { + if message.OpCode != ws.OpBinary { + logger.Info("receive message type: ", message.OpCode) + continue + } + + // decode payload + payload, err := protocol.DecodeSo(message.Payload) + if err != nil { + logger.Errorf("decode message error: len=%d", len(message.Payload), err) + return gnet.Close + } + + header := payload.Header + service, method := header.Svc, header.Target + // authorization + if !authorized { + if !(service == "Auth" && method == "Verify") { + writeError(conn, header.SeqId, session.UnauthorizedRequestError.Error()) + return gnet.Close + } + } + + // payload -> processing channel + select { + case wss.requests <- payload: + default: + writeError(conn, header.SeqId, "server busy") + } + //msgLen := len(message.Payload) + //if msgLen > 128 { + // logging.Infof("conn[%v] receive [op=%v] [msg=%v..., len=%d]", c.RemoteAddr().String(), message.OpCode, string(message.Payload[:128]), len(message.Payload)) + //} else { + // logging.Infof("conn[%v] receive [op=%v] [msg=%v, len=%d]", c.RemoteAddr().String(), message.OpCode, string(message.Payload), len(message.Payload)) + //} + //// This is the echo server + //err = wsutil.WriteServerMessage(c, message.OpCode, message.Payload) + //if err != nil { + // logging.Infof("conn[%v] [err=%v]", c.RemoteAddr().String(), err.Error()) + // return gnet.Close + //} + } + return gnet.None +} + +func (wss *GnetWsServer) OnTick() (delay time.Duration, action gnet.Action) { + logging.Infof("[connected-count=%v]", atomic.LoadInt64(&wss.connected)) + return 10 * time.Second, gnet.None +} + +func writeError(conn session.NetConn, seqId int32, errMsg string) { + header := &protocol.Header{} + payload := &protocol.Payload{Header: header} + + header.Magic = protocol.Magic + header.Type = protocol.TypeError + header.Status = 50 + header.SeqId = seqId + payload.Body = []byte(errMsg) + message, err := protocol.EncodeSo(payload) + if err != nil { + logger.Error("NetClient writeError encode error: ", err) + return + } + err = conn.Write(message) + if err != nil { + logger.Error("net conn write error msg fail: ", err) + return + } + return +} + +func main() { + var port int + var multicore bool + + // Example command: go run main.go --port 8080 --multicore=true + flag.IntVar(&port, "port", 9080, "server port") + flag.BoolVar(&multicore, "multicore", true, "multicore") + flag.Parse() + + wss := &GnetWsServer{addr: fmt.Sprintf("tcp://127.0.0.1:%d", port), multicore: multicore} + + // Start serving! + log.Println("server exits:", gnet.Run(wss, wss.addr, gnet.WithMulticore(multicore), gnet.WithReusePort(true), gnet.WithTicker(true))) +} diff --git a/internal/gateway_ws/gws_server/conn_handler.go b/internal/gateway_ws/gws_server/conn_handler.go index 4f8320d..6aac344 100644 --- a/internal/gateway_ws/gws_server/conn_handler.go +++ b/internal/gateway_ws/gws_server/conn_handler.go @@ -23,19 +23,17 @@ const ( ) type GwsHandler struct { - postalServerAddress string - grpcFactory *generic.GrpcGenericClientFactory - sessionStore session.Store // 当前连接用户,内存缓存 + grpcFactory *generic.GrpcGenericClientFactory + sessionStore session.Store // 当前连接用户,内存缓存 } -func NewGwsHandler(postalServerAddress string, +func NewGwsHandler( grpcFactory *generic.GrpcGenericClientFactory, sessionStore session.Store, ) *GwsHandler { return &GwsHandler{ - grpcFactory: grpcFactory, - postalServerAddress: postalServerAddress, - sessionStore: sessionStore, + grpcFactory: grpcFactory, + sessionStore: sessionStore, } }