package ws_handler import ( "common/log" "common/net/socket" "context" "fmt" "go.uber.org/zap" "runtime/debug" "sync" "time" ) var GatewaySID int64 type Client struct { sync.WaitGroup conn socket.ISocketConn // Socket mailChan chan Event // 邮箱队列 logger *zap.SugaredLogger // 日志 ctx context.Context // 上下文 cancel context.CancelFunc // 取消上下文 heartBeat time.Time // 最后一次心跳 USN int64 // 用户ID SceneSID int64 // 场景服ID InstanceID int32 // 副本ID,副本类型 UniqueNo int64 // 副本唯一编号 } func NewClient(usn int64, conn socket.ISocketConn) *Client { client := &Client{ USN: usn, conn: conn, logger: log.GetLogger().Named(fmt.Sprintf("usn:%v", usn)), heartBeat: time.Now(), mailChan: make(chan Event, 1024), } client.ctx, client.cancel = context.WithCancel(context.Background()) client.Add(1) go client.Loop() return client } 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) > 60*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") } } // CloseClient 关闭客户端(同步,会等待onClose执行完成) func (c *Client) CloseClient() { if c.cancel != nil { c.cancel() c.Wait() } } func (c *Client) onClose() { if c.conn != nil { _ = c.conn.Close() c.conn = nil } if c.mailChan != nil { close(c.mailChan) c.mailChan = nil } UserMgr.Delete(c.USN) c.onLeave() c.Done() }