package server import ( "dlt645-server/protocol" "errors" "github.com/gogf/gf/encoding/gbinary" "github.com/gogf/gf/frame/g" "github.com/gogf/gf/net/gtcp" "github.com/gogf/gf/os/glog" "io" "net" "strings" "syscall" "time" ) type Client struct { Id string Address []byte srv *Server conn *gtcp.Conn sendChan chan []byte closeChan chan struct{} regHandler func(id string, c *Client) closeHandler func(id string, c *Client) lastHeartBeat time.Time gatewayId uint16 isReg bool packetContext *protocol.PacketContext } func NewClient(s *Server, conn *gtcp.Conn) *Client { return &Client{ srv: s, conn: conn, sendChan: make(chan []byte), closeChan: make(chan struct{}), packetContext: new(protocol.PacketContext), } } func (c *Client) SendLoop() { for { select { case buf := <-c.sendChan: err := c.send(buf) if err != nil { glog.Errorf("指令发送失败:%s", err.Error()) continue } timer := time.NewTimer(5 * time.Second) for { select { case <-timer.C: glog.Errorf("接收指令超时") break default: var code string receiveBuf, err := c.conn.Recv(-1) if err != nil { c.readError(err) break } if len(receiveBuf) > 0 { data, err := c.srv.message.Decode(c.packetContext, receiveBuf) if err != nil { glog.Errorf("解析报文失败:%s", err.Error()) break } var reportData interface{} switch data.DataType { case protocol.IsPower: reportData = protocol.PowerData{ ActivePower: data.ActivePower, } code = "power" case protocol.IsVData: reportData = protocol.VIData{ AV: data.AV, BV: data.BV, CV: data.CV, } code = "voltage" case protocol.IsIData: reportData = protocol.IData{ AI: data.AI, BI: data.BI, CI: data.CI, } code = "current" default: break } if reportData != nil { if err := c.srv.ReportStatus(c.Id, reportData, code); err != nil { glog.Errorf("数据上报发送错误:%s", err.Error()) } } } } break } } } } func (c *Client) SetId(id string) { c.Id = id } func (c *Client) SetAddress(address []byte) { c.Address = address } func (c *Client) readError(err error) { defer c.closeConnection() if err == io.EOF || isErrConnReset(err) { return } glog.Errorf("读取数据发生错误:%s", err.Error()) } func (c *Client) closeConnection() { _ = c.conn.Close() c.conn = nil close(c.closeChan) c.SetId("") c.isReg = false if c.closeHandler != nil { c.closeHandler(c.Id, c) } } // isErrConnReset read: connection reset by peer func isErrConnReset(err error) bool { if ne, ok := err.(*net.OpError); ok { return strings.Contains(ne.Err.Error(), syscall.ECONNRESET.Error()) } return false } func (c *Client) send(buf []byte) error { if c.conn == nil { return nil } err := c.conn.Send(buf) if err != nil { glog.Error(err) c.closeConnection() return err } glog.Debugf("指令发送成功:%2X", buf) return nil } func (c *Client) GetActivePower() { for { entity := protocol.Dlt_0x33333433{} sendByte, _ := entity.Encode(c.packetContext) if err := c.writeSensChan(sendByte); err != nil { continue } time.Sleep(time.Duration(g.Cfg().GetInt("Server.PowerFrequency")) * time.Second) } } func (c *Client) writeSensChan(buf []byte) error { timer := time.NewTimer(3 * time.Second) select { case <-timer.C: glog.Debugf("write timeout--------------") return errors.New("write timeout") default: c.sendChan <- buf } return nil } func (c *Client) GetVBlock() { for { entity := protocol.Dlt_0x33323435{} sendByte, _ := entity.Encode(c.packetContext) if err := c.writeSensChan(sendByte); err != nil { continue } time.Sleep(time.Duration(g.Cfg().GetInt("Server.VIFrequency")) * time.Second) } } func (c *Client) GetIBlock() { for { entity := protocol.Dlt_0x33323535{} sendByte, _ := entity.Encode(c.packetContext) if err := c.writeSensChan(sendByte); err != nil { continue } time.Sleep(time.Duration(g.Cfg().GetInt("Server.VIFrequency")) * time.Second) } } func (c *Client) SendGetAddress() error { entity := new(protocol.Dlt_0x93) sendBuf, _ := entity.Encode(c.packetContext) err := c.send(sendBuf) if err != nil { return err } timer := time.NewTimer(10 * time.Second) for { select { case <-timer.C: glog.Errorf("读取数据超时") return errors.New("读取数据超时") default: receiveBuf, err := c.conn.Recv(-1) if err != nil { c.readError(err) return err } glog.Debugf("收到数据:%2X", receiveBuf) if len(receiveBuf) > 0 { if !c.isReg { c.SetId(gbinary.DecodeToString(receiveBuf)) c.isReg = true if c.regHandler != nil { c.regHandler(c.Id, c) } break } _, err = c.srv.message.Decode(c.packetContext, receiveBuf) if err != nil { glog.Errorf("解析报文失败:%s", err.Error()) return err } c.SetId(c.packetContext.GetDeviceId()) return nil } } } }