feat(common): 添加 WebSocket 支持
- 在 ClientData 结构中添加 WsCodec 字段 - 实现 WebSocket 升级和消息处理逻辑 - 添加 WebSocket 相关的依赖包
This commit is contained in:
@@ -7,6 +7,8 @@ type ClientData struct {
|
||||
player *Player //客户实体
|
||||
//UserID uint32
|
||||
m sync.Mutex
|
||||
|
||||
wsmsg WsCodec
|
||||
}
|
||||
|
||||
func (cd *ClientData) SetPlayer(player *Player) {
|
||||
@@ -19,6 +21,11 @@ func (cd *ClientData) GetPlayer() *Player {
|
||||
defer cd.m.Unlock()
|
||||
return cd.player
|
||||
}
|
||||
func (cd *ClientData) Getwsmsg() *WsCodec {
|
||||
cd.m.Lock()
|
||||
defer cd.m.Unlock()
|
||||
return &cd.wsmsg
|
||||
}
|
||||
func (cd *ClientData) SetCrossDomain(isCrossDomain bool) {
|
||||
cd.m.Lock()
|
||||
defer cd.m.Unlock()
|
||||
@@ -35,6 +42,7 @@ func NewClientData() *ClientData {
|
||||
isCrossDomain: false,
|
||||
player: nil,
|
||||
m: sync.Mutex{},
|
||||
wsmsg: WsCodec{},
|
||||
}
|
||||
return &cd
|
||||
|
||||
|
||||
161
common/data/entity/wscodec.go
Normal file
161
common/data/entity/wscodec.go
Normal file
@@ -0,0 +1,161 @@
|
||||
package entity
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/gobwas/ws"
|
||||
"github.com/gobwas/ws/wsutil"
|
||||
"github.com/panjf2000/gnet/v2"
|
||||
"github.com/panjf2000/gnet/v2/pkg/logging"
|
||||
)
|
||||
|
||||
type WsCodec struct {
|
||||
upgraded bool // 链接是否升级
|
||||
Buf bytes.Buffer // 从实际socket中读取到的数据缓存
|
||||
wsMsgBuf wsMessageBuf // ws 消息缓存
|
||||
Isinitws bool
|
||||
}
|
||||
|
||||
type wsMessageBuf struct {
|
||||
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 || errors.Is(err, io.ErrUnexpectedEOF) { //数据不完整，不跳过 buf 中的 skipN 字节（此时 buf 中存放的仅是部分 "handshake data" bytes），下次再尝试读取
|
||||
return
|
||||
}
|
||||
buf.Next(skipN)
|
||||
logging.Errorf("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)
|
||||
|
||||
ok = true
|
||||
w.upgraded = true
|
||||
return
|
||||
}
|
||||
func (w *WsCodec) ReadBufferBytes(c gnet.Conn) (gnet.Action, int) {
|
||||
size := c.InboundBuffered()
|
||||
//buf := make([]byte, size)
|
||||
read, err := c.Peek(size)
|
||||
if err != nil {
|
||||
logging.Errorf("read err! %v", err)
|
||||
return gnet.Close, 0
|
||||
}
|
||||
// if read < size {
|
||||
// logging.Errorf("read bytes len err! size: %d read: %d", size, read)
|
||||
// return gnet.Close
|
||||
// }
|
||||
w.Buf.Write(read)
|
||||
return gnet.None, size
|
||||
}
|
||||
func (w *WsCodec) Decode(c gnet.Conn) (outs []wsutil.Message, err error) {
|
||||
fmt.Println("do Decode")
|
||||
messages, err := w.readWsMessages()
|
||||
if err != nil {
|
||||
logging.Errorf("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 {
|
||||
// 从 in 中读出 header，并将 header bytes 写入 msgBuf.cachedBuf
|
||||
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 || errors.Is(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)
|
||||
// 从 in 中读出 data，并将 data bytes 写入 msgBuf.cachedBuf
|
||||
if dataLen > 0 {
|
||||
if in.Len() < dataLen { //数据不完整
|
||||
fmt.Println(in.Len(), dataLen)
|
||||
logging.Infof("incomplete data")
|
||||
return
|
||||
}
|
||||
|
||||
_, err = io.CopyN(&msgBuf.cachedBuf, in, int64(dataLen))
|
||||
if err != nil {
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -25,8 +25,13 @@ require (
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||
github.com/go-redis/redis/v8 v8.11.5 // indirect
|
||||
github.com/gobwas/httphead v0.1.0 // indirect
|
||||
github.com/gobwas/pool v0.2.1 // indirect
|
||||
github.com/gobwas/ws v1.4.0 // indirect
|
||||
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
|
||||
github.com/golang/protobuf v1.5.3 // indirect
|
||||
github.com/kavu/go_reuseport v1.5.0 // indirect
|
||||
github.com/libp2p/go-reuseport v0.4.0 // indirect
|
||||
github.com/pointernil/bitset32 v0.0.1 // indirect
|
||||
github.com/yitter/idgenerator-go v1.3.3 // indirect
|
||||
google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d // indirect
|
||||
|
||||
@@ -32,6 +32,12 @@ github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ4
|
||||
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
|
||||
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
|
||||
github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
|
||||
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.4.0 h1:CTaoG1tojrh4ucGPcoJFiAQUAsEWekEWvLy7GsVNqGs=
|
||||
github.com/gobwas/ws v1.4.0/go.mod h1:G3gNqMNtPppf5XUz7O4shetPpcZ1VJ7zt18dlUeakrc=
|
||||
github.com/gogf/gf/contrib/nosql/redis/v2 v2.6.3/go.mod h1:2+evGu1xAlamaYuDdSqa7QCiwPTm1RrGsUFSMc8PyLc=
|
||||
github.com/gogf/gf/v2 v2.6.3 h1:DoqeuwU98wotpFoDSQEx8RZbmJdK8KdGiJtzJeqpyIo=
|
||||
github.com/gogf/gf/v2 v2.6.3/go.mod h1:x2XONYcI4hRQ/4gMNbWHmZrNzSEIg20s2NULbzom5k0=
|
||||
@@ -52,9 +58,13 @@ github.com/grokify/html-strip-tags-go v0.0.1 h1:0fThFwLbW7P/kOiTBs03FsJSV9RM2M/Q
|
||||
github.com/grokify/html-strip-tags-go v0.0.1/go.mod h1:2Su6romC5/1VXOQMaWL2yb618ARB8iVo6/DR99A6d78=
|
||||
github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 h1:X5VWvz21y3gzm9Nw/kaUeku/1+uBhcekkmy4IkffJww=
|
||||
github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1/go.mod h1:Zanoh4+gvIgluNqcfMVTJueD4wSS5hT7zTt4Mrutd90=
|
||||
github.com/kavu/go_reuseport v1.5.0 h1:UNuiY2OblcqAtVDE8Gsg1kZz8zbBWg907sP1ceBV+bk=
|
||||
github.com/kavu/go_reuseport v1.5.0/go.mod h1:CG8Ee7ceMFSMnx/xr25Vm0qXaj2Z4i5PWoUx+JZ5/CU=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/libp2p/go-reuseport v0.4.0 h1:nR5KU7hD0WxXCJbmw7r2rhRYruNRl2koHw8fQscQm2s=
|
||||
github.com/libp2p/go-reuseport v0.4.0/go.mod h1:ZtI03j/wO5hZVDFo2jKywN6bYKWLOy8Se6DrI2E1cLU=
|
||||
github.com/lunixbochs/struc v0.0.0-20241101090106-8d528fa2c543 h1:GxMuVb9tJajC1QpbQwYNY1ZAo1EIE8I+UclBjOfjz/M=
|
||||
github.com/lunixbochs/struc v0.0.0-20241101090106-8d528fa2c543/go.mod h1:vy1vK6wD6j7xX6O6hXe621WabdtNkou2h7uRtTfRMyg=
|
||||
github.com/magiconair/properties v1.8.6 h1:5ibWZ6iY0NctNGWo87LalDlEZ6R41TqbbDamhfG/Qzo=
|
||||
|
||||
@@ -2,21 +2,26 @@ package socket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"blazing/common/data/entity"
|
||||
|
||||
"github.com/gobwas/ws/wsutil"
|
||||
"github.com/gogf/gf/v2/os/glog"
|
||||
"github.com/panjf2000/gnet/v2"
|
||||
"github.com/panjf2000/gnet/v2/pkg/logging"
|
||||
)
|
||||
|
||||
func (s *Server) Boot() error {
|
||||
// go s.bootws()
|
||||
err := gnet.Run(s, s.network+"://"+s.addr,
|
||||
gnet.WithMulticore(true),
|
||||
gnet.WithTicker(true),
|
||||
// gnet.WithReusePort(true),
|
||||
// gnet.WithReuseAddr(true),
|
||||
gnet.WithSocketRecvBuffer(s.bufferSize))
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -64,19 +69,60 @@ func (s *Server) OnBoot(eng gnet.Engine) gnet.Action {
|
||||
return gnet.None
|
||||
}
|
||||
|
||||
func (s *Server) OnOpen(_ gnet.Conn) (out []byte, action gnet.Action) {
|
||||
func (s *Server) OnOpen(conn gnet.Conn) (out []byte, action gnet.Action) {
|
||||
if conn.Context() == nil {
|
||||
conn.SetContext(entity.NewClientData()) //注入data
|
||||
}
|
||||
|
||||
atomic.AddInt64(&s.connected, 1)
|
||||
|
||||
return nil, gnet.None
|
||||
}
|
||||
|
||||
func (s *Server) OnTraffic(conn gnet.Conn) (action gnet.Action) {
|
||||
if conn.Context() == nil {
|
||||
conn.SetContext(entity.NewClientData()) //注入data
|
||||
func (s *Server) OnTraffic(c gnet.Conn) (action gnet.Action) {
|
||||
if s.network != "tcp" {
|
||||
return gnet.Close
|
||||
}
|
||||
|
||||
if s.network == "tcp" {
|
||||
return s.handleTcp(conn)
|
||||
ws := c.Context().(*entity.ClientData).Getwsmsg()
|
||||
tt, len1 := ws.ReadBufferBytes(c)
|
||||
if tt == gnet.Close {
|
||||
|
||||
return gnet.Close
|
||||
}
|
||||
|
||||
ok, action := ws.Upgrade(c)
|
||||
if !ok {
|
||||
s.handleTcp(c)
|
||||
return gnet.None
|
||||
} else {
|
||||
fmt.Println(ws.Buf.Bytes())
|
||||
c.Read(make([]byte, len1))
|
||||
}
|
||||
|
||||
if ws.Buf.Len() <= 0 {
|
||||
return gnet.None
|
||||
}
|
||||
messages, err := ws.Decode(c)
|
||||
if err != nil {
|
||||
return gnet.Close
|
||||
}
|
||||
if messages == nil {
|
||||
return
|
||||
}
|
||||
for _, message := range messages {
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user