Compare commits
No commits in common. 'aa943bd7619afccb5f86a173dcbf6e8b896a56cc' and 'bb6e22c4e01618c4e0c999ff20a034e70c6e26ad' have entirely different histories.
aa943bd761
...
bb6e22c4e0
26 changed files with 256 additions and 1042 deletions
@ -1,161 +0,0 @@
|
||||
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 |
||||
} |
||||
} |
||||
@ -1,23 +0,0 @@
|
||||
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) |
||||
} |
||||
@ -1,324 +0,0 @@
|
||||
package gnet_server |
||||
|
||||
import ( |
||||
"context" |
||||
"fmt" |
||||
"github.com/gobwas/ws" |
||||
"google.golang.org/protobuf/proto" |
||||
"regexp" |
||||
"sonet/api/gen/auth" |
||||
"sonet/api/gen/postal" |
||||
"sonet/pkg/grpc/generic" |
||||
"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 gnetRequest struct { |
||||
conn *gnetConn |
||||
payload *protocol.Payload |
||||
} |
||||
|
||||
type GnetWsServer struct { |
||||
gnet.BuiltinEventEngine |
||||
|
||||
addr string |
||||
multicore bool |
||||
eng gnet.Engine |
||||
connected int64 |
||||
|
||||
grpcFactory *generic.GrpcGenericClientFactory |
||||
sessionStore session.Store // <string, *session.NetSubject> 当前连接用户,内存缓存
|
||||
requests chan *gnetRequest |
||||
} |
||||
|
||||
func NewGnetWsServer( |
||||
grpcFactory *generic.GrpcGenericClientFactory, |
||||
sessionStore session.Store) *GnetWsServer { |
||||
return &GnetWsServer{ |
||||
grpcFactory: grpcFactory, |
||||
sessionStore: sessionStore, |
||||
} |
||||
} |
||||
|
||||
func (wss *GnetWsServer) Init(ctx context.Context) { |
||||
wss.requests = make(chan *gnetRequest, 1024*16) |
||||
|
||||
// todo 针对不同服务的协程池
|
||||
for i := 0; i < 32; i++ { |
||||
go wss.dispatch(ctx) |
||||
} |
||||
} |
||||
|
||||
// dispatch 多个 goroutine 进行请求 dispatch
|
||||
func (wss *GnetWsServer) dispatch(ctx context.Context) { |
||||
for { |
||||
select { |
||||
case <-ctx.Done(): |
||||
return |
||||
case request := <-wss.requests: |
||||
func() { |
||||
conn, payload := request.conn, request.payload |
||||
header := payload.Header |
||||
service, method := header.Svc, header.Target |
||||
|
||||
// grpc generic call
|
||||
ctx := context.Background() |
||||
ctx2, cancel2 := context.WithTimeout(ctx, time.Second*3) |
||||
defer cancel2() |
||||
grpcClient, err := wss.grpcFactory.GetClient(ctx2, service) |
||||
if err != nil { |
||||
logger.Error("get grpc generic client error: ", err) |
||||
writeError(conn, header.SeqId, fmt.Sprintf("service %s not avaliable: %s", header.Svc, err.Error())) |
||||
return |
||||
} |
||||
|
||||
// put grpc request session
|
||||
if conn.uid != "" { |
||||
ctx = session.PutSubject(ctx, session.NewRpcSubject(conn.uid)) |
||||
} |
||||
ctx3, cancel3 := context.WithTimeout(ctx, time.Second*10) |
||||
defer cancel3() |
||||
resp, err := grpcClient.InvokeUnary(ctx3, header.Target, payload.Body) |
||||
if err != nil { |
||||
writeError(conn, header.SeqId, unwrapRpcError(err.Error())) |
||||
logger.Error("grpc generic call error: ", err) |
||||
return |
||||
} |
||||
|
||||
// write response
|
||||
if resp == nil { // proto.Empty
|
||||
header.Target = "" |
||||
} else { |
||||
payload.Body, err = resp.Marshal() |
||||
if err != nil { |
||||
logger.Error("generic call response marshal error: ", err) |
||||
writeError(conn, header.SeqId, "server error") |
||||
return |
||||
} |
||||
header.Target = resp.XXX_MessageName() |
||||
} |
||||
|
||||
// auth verify success
|
||||
if service == "Auth" && method == "Verify" { |
||||
var channel *session.Channel |
||||
conn.uid, channel, err = wss.extractAuthVerify(conn, payload.Body) |
||||
if err != nil { |
||||
logger.Error("extractAuthVerify error: ", err) |
||||
writeError(conn, header.SeqId, "server error") |
||||
return |
||||
} |
||||
// channel
|
||||
go wss.processWriting(channel) |
||||
} |
||||
|
||||
header.Type = protocol.TypeResponse |
||||
resMessage, err := protocol.EncodeSo(payload) |
||||
if err != nil { |
||||
logger.Error("grpc generic call error: ", err) |
||||
return |
||||
} |
||||
err = conn.Write(resMessage) |
||||
if err != nil { |
||||
logger.Error("gws write response error: ", err) |
||||
return |
||||
} |
||||
}() |
||||
} |
||||
} |
||||
} |
||||
|
||||
func encodeDeliverMessage(msg *postal.Message) (bytes []byte, err error) { |
||||
header := &protocol.Header{ |
||||
Magic: protocol.Magic, |
||||
Type: protocol.TypeNotice, |
||||
UrlType: 1, |
||||
SerializeType: 1, |
||||
Svc: msg.Svc, |
||||
Target: msg.Msg, |
||||
} |
||||
payload := &protocol.Payload{Header: header, Body: msg.Body} |
||||
bytes, err = protocol.EncodeSo(payload) |
||||
if err != nil { |
||||
logger.Error("encode deliver message error: ", err) |
||||
} |
||||
return |
||||
} |
||||
|
||||
// todo close
|
||||
func (c *GnetWsServer) processWriting(channel *session.Channel) { |
||||
for { |
||||
msg := channel.Ready() |
||||
bytes, err := encodeDeliverMessage(msg) |
||||
if err != nil { |
||||
logger.Error("encode dispatch message error:", err) |
||||
continue |
||||
} |
||||
err = channel.Conn.Write(bytes) |
||||
if err != nil { |
||||
logger.Error("write dispatch message error:", err) |
||||
} |
||||
} |
||||
} |
||||
|
||||
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
|
||||
request := &gnetRequest{conn: conn, payload: payload} |
||||
select { |
||||
case wss.requests <- request: |
||||
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 (c *GnetWsServer) extractAuthVerify(conn *gnetConn, resp []byte) (sessionUid string, channel *session.Channel, err error) { |
||||
authSubject := &auth.Subject{} |
||||
err = proto.Unmarshal(resp, authSubject) |
||||
if err != nil { |
||||
logger.Error("grpc generic call error: ", err) |
||||
return |
||||
} |
||||
// 设置连接身份信息
|
||||
sessionUid = authSubject.Uid |
||||
|
||||
// 关闭旧的链接
|
||||
oldChannel, ok := c.sessionStore.Load(sessionUid) |
||||
if ok { |
||||
if err = oldChannel.Conn.(*gnetConn).conn.Close(); err != nil { |
||||
logger.Error("close gnet conn error: ", err) |
||||
} |
||||
logger.Infof("close subject old conn: %s\n", sessionUid) |
||||
} |
||||
// 存储session到内存
|
||||
channel = session.NewChannel(sessionUid, conn) |
||||
c.sessionStore.Store(sessionUid, channel) |
||||
logger.Infof("subject online: %s\n", sessionUid) |
||||
return |
||||
} |
||||
|
||||
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 |
||||
} |
||||
|
||||
var reg = regexp.MustCompile("^rpc error:.*?desc ?= ?(.*)$") |
||||
|
||||
// unwrapRpcError 提取 grpc err: fmt.Sprintf("rpc error: code = %s desc = %s", s.Code(), s.Message())
|
||||
func unwrapRpcError(err string) string { |
||||
finds := reg.FindStringSubmatch(err) |
||||
if len(finds) > 1 { |
||||
return finds[1] |
||||
} |
||||
return err |
||||
} |
||||
Loading…
Reference in new issue