网络层

Server/Gateway/handler/ws_handler/client.go

@@ -3,7 +3,9 @@ package ws_handler
 import (
 	"common/log"
 	"common/net/socket"
+	"common/utils"
 	"context"
+	"fmt"
 	"go.uber.org/zap"
 	"runtime/debug"
 	"sync"
@@ -14,29 +16,90 @@ type Client struct {
 	sync.WaitGroup
 	conn      socket.ISocketConn // Socket
 	mailChan  chan Event         // 邮箱队列
-	logger    *zap.SugaredLogger
-	ctx       context.Context
-	cancel    context.CancelFunc
-	heartBeat time.Time
+	logger    *zap.SugaredLogger // 日志
+	ctx       context.Context    // 上下文
+	cancel    context.CancelFunc // 取消上下文
+	heartBeat time.Time          // 最后一次心跳
 
 	UID int32
 }
 
 func NewClient(uid int32, conn socket.ISocketConn) *Client {
-	client := &Client{}
-	client.UID = uid
-	client.conn = conn
-	client.logger = log.GetLogger().Named("uid").With("uid", client.UID)
-	client.logger.Errorf("错误日志 %v", 1)
-
-	client.heartBeat = time.Now()
-	client.mailChan = make(chan Event, 1024)
+	client := &Client{
+		UID:       uid,
+		conn:      conn,
+		logger:    log.GetLogger().Named(fmt.Sprintf("uid:%v", uid)),
+		heartBeat: time.Now(),
+		mailChan:  make(chan Event, 1024),
+	}
 	client.ctx, client.cancel = context.WithCancel(context.Background())
+	client.Add(1)
 	go client.Loop()
 	return client
 }
 
-// CloseClient 关闭客户端
+func (c *Client) Loop() {
+	defer func() {
+		if err := recover(); err != nil {
+			c.logger.Errorf("Client Loop err: %v", err)
+			debug.PrintStack()
+		}
+	}()
+	defer c.onClose()
+	t := time.NewTicker(20 * time.Second)
+	defer t.Stop()
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case evt, ok := <-c.mailChan:
+			if ok {
+				c.handle(evt)
+			}
+		case <-t.C:
+			_ = c.conn.Ping()
+			if time.Now().Sub(c.heartBeat) > 120*time.Second {
+				return
+			}
+		}
+	}
+}
+
+func (c *Client) OnEvent(event Event) {
+	defer func() {
+		if err := recover(); err != nil {
+			c.logger.Warnf(fmt.Sprintf("send event chan error: %v", err))
+		}
+	}()
+	select {
+	case c.mailChan <- event:
+	default:
+		c.logger.Warnf("Client mailChan full")
+	}
+}
+
+func (c *Client) handle(event Event) {
+	switch e := event.(type) {
+	case *ClientEvent:
+		m, err := parseMsg(e.Msg)
+		if err != nil {
+			c.logger.Errorf("handle event json.Unmarshal err: %v", err)
+			c.cancel()
+		}
+		c.logger.Infof("收到客户端消息：%+v", *m)
+		switch m.Type {
+		case "init":
+			_ = c.conn.Write(wapMsg(&msg{
+				Type: "init",
+				Data: fmt.Sprintf("[%v,%v]", utils.RandInt(1, 100), utils.RandInt(1, 100)),
+			}))
+		}
+	case *PongEvent:
+		c.heartBeat = time.Now()
+	}
+}
+
+// CloseClient 关闭客户端（同步，会等待onClose执行完成）
 func (c *Client) CloseClient() {
 	if c.cancel != nil {
 		c.cancel()
@@ -56,34 +119,3 @@ func (c *Client) onClose() {
 	UserMgr.Delete(c.UID)
 	c.Done()
 }
-
-func (c *Client) Loop() {
-	defer func() {
-		if err := recover(); err != nil {
-			c.logger.Errorf("Client Loop err: %v", err)
-			debug.PrintStack()
-		}
-	}()
-	defer c.onClose()
-	c.Add(1)
-	//心跳检测
-	hearBeatTicker := time.NewTicker(3000 * time.Millisecond)
-	for {
-		select {
-		case <-c.ctx.Done():
-			return
-		case _, _ = <-c.mailChan:
-
-		case <-hearBeatTicker.C:
-			// 心跳超时直接关掉连接
-			if c.checkHeartBeatTimeout() {
-				return
-			}
-		}
-	}
-}
-
-func (c *Client) checkHeartBeatTimeout() bool {
-	sub := time.Now().Sub(c.heartBeat)
-	return sub > 60*time.Second
-}