feat 优化网络库

Server/common/net/socket/websocket/websocket.go

@@ -12,7 +12,6 @@ import (
 
 // WSServer 实现GNet库接口
 type WSServer struct {
-	gnet.BuiltinEventEngine
 	eng            gnet.Engine
 	i              socket.ISocketServer
 	logger         logging.Logger // 日志
@@ -20,29 +19,31 @@ type WSServer struct {
 	unUpgradeConn  sync.Map
 }
 
-func NewWSServer(i socket.ISocketServer, logger logging.Logger, timeout time.Duration) *WSServer {
+func NewWSServer(i socket.ISocketServer, logger logging.Logger, upgradeTimeout time.Duration) *WSServer {
 	if i == nil {
 		return nil
 	}
 	return &WSServer{
 		i:              i,
 		logger:         logger,
-		upgradeTimeout: timeout,
+		upgradeTimeout: upgradeTimeout,
 		unUpgradeConn:  sync.Map{},
 	}
 }
 
-func (s *WSServer) Run(logger logging.Logger, addr string, multiCore, reusePort, tick, lockOSThread, reuseAddr bool, processNum int) error {
+func (s *WSServer) Run(addr string, multicore bool, numEventLoop int, tcpNoDelay gnet.TCPSocketOpt, readBufferCap, writeBufferCap int, lockOSThread bool, reusePort, ticker bool, logger logging.Logger) error {
 	return gnet.Run(
 		s,
 		addr,
-		gnet.WithMulticore(multiCore),
-		gnet.WithNumEventLoop(processNum),
-		gnet.WithReusePort(reusePort),
-		gnet.WithTicker(tick),
-		gnet.WithLogger(logger),
+		gnet.WithMulticore(multicore),
+		gnet.WithNumEventLoop(numEventLoop),
+		gnet.WithTCPNoDelay(tcpNoDelay),
+		gnet.WithReadBufferCap(readBufferCap),
+		gnet.WithWriteBufferCap(writeBufferCap),
 		gnet.WithLockOSThread(lockOSThread),
-		gnet.WithReuseAddr(reuseAddr),
+		gnet.WithReusePort(reusePort),
+		gnet.WithTicker(ticker),
+		gnet.WithLogger(logger),
 	)
 }
 
@@ -55,6 +56,9 @@ func (s *WSServer) OnBoot(eng gnet.Engine) gnet.Action {
 	return gnet.None
 }
 
+func (s *WSServer) OnShutdown(_ gnet.Engine) {
+}
+
 func (s *WSServer) OnOpen(c gnet.Conn) ([]byte, gnet.Action) {
 	ws := &WSConn{
 		Conn:      c,
@@ -89,67 +93,43 @@ func (s *WSServer) OnClose(c gnet.Conn, err error) (action gnet.Action) {
 }
 
 // OnTraffic fires when a local socket receives data from the peer.
-func (s *WSServer) OnTraffic(c gnet.Conn) (action gnet.Action) {
-	tmp := c.Context()
-	if tmp == nil {
-		s.logger.Errorf("OnTraffic context nil: %v", c)
-		action = gnet.Close
-		return
-	}
-
-	ws, ok := tmp.(*WSConn)
-	if !ok {
-		ws.logger.Errorf("OnTraffic convert ws error: %v", tmp)
-		action = gnet.Close
-		return
-	}
-
-	action = ws.readBytesBuf(c)
-	if action != gnet.None {
-		return
+func (s *WSServer) OnTraffic(c gnet.Conn) gnet.Action {
+	ws := c.Context().(*WSConn)
+	if action := ws.readBytesBuf(c); action != gnet.None {
+		return action
 	}
 
 	if !ws.isUpgrade {
-		var data []byte
-		data, ok, action = ws.upgrade()
-		if ok {
+		data, action := ws.upgrade()
+		if len(data) > 0 {
 			s.unUpgradeConn.Delete(c.RemoteAddr().String())
-			s.i.OnHandShake(ws)
-			if data != nil {
-				err := ws.Conn.AsyncWrite(data, nil)
-				if err != nil {
-					ws.logger.Errorf("update ws write upgrade protocol error", err)
-					action = gnet.Close
-				}
-			}
-		}
-	} else {
-		msg, err := ws.readWsMessages()
-		if err != nil {
-			ws.logger.Errorf("read ws messages errors", err)
-			return gnet.Close
+			action = gnet.Action(s.i.OnHandShake(ws, data, s.OnHandShakeFinish))
 		}
+		return action
+	}
 
-		if msg != nil {
-			for _, m := range msg {
-				if socket.OpCode(m.OpCode) == socket.OpPong {
-					s.i.OnPong(ws)
-					continue
-				}
-				if socket.OpCode(m.OpCode) == socket.OpClose {
-					return gnet.Close
-				}
-				if socket.OpCode(m.OpCode) == socket.OpPing {
-					continue
-				}
-				a := s.i.OnMessage(ws, m.Payload)
-				if gnet.Action(a) != gnet.None {
-					action = gnet.Action(a)
-				}
+	if msg, err := ws.readWsMessages(); err != nil {
+		ws.logger.Errorf("read ws messages err: %v", err)
+		return gnet.Close
+	} else if msg != nil {
+		for _, m := range msg {
+			if socket.OpCode(m.OpCode) == socket.OpPong {
+				s.i.OnPong(ws)
+				continue
+			}
+			if socket.OpCode(m.OpCode) == socket.OpClose {
+				return gnet.Close
+			}
+			if socket.OpCode(m.OpCode) == socket.OpPing {
+				continue
+			}
+			a := s.i.OnMessage(ws, m.Payload)
+			if gnet.Action(a) != gnet.None {
+				return gnet.Action(a)
 			}
 		}
 	}
-	return
+	return gnet.None
 }
 
 // OnTick fires immediately after the engine starts and will fire again
@@ -194,3 +174,14 @@ func (s *WSServer) OnTick() (delay time.Duration, action gnet.Action) {
 
 	return
 }
+
+// OnHandShakeFinish 握手完成
+func (s *WSServer) OnHandShakeFinish(conn socket.ISocketConn, hsResp []byte) {
+	ws := conn.(*WSConn)
+	if err := ws.Conn.AsyncWrite(hsResp, nil); err != nil {
+		ws.logger.Errorf("OnHandShakeFinish err: %v", err)
+		if err = ws.Close(); err != nil {
+			ws.logger.Errorf("OnHandShakeFinish Close error: %v", err)
+		}
+	}
+}