package server import ( "dlt645-server/protocol" "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{} closeHandler func(id string, c *Client) lastHeartBeat time.Time done chan struct{} gatewayId uint16 isReg bool } func (c *Client) ReadLoop(ctx *protocol.PacketContext) { defer c.srv.grWG.Done() for { buf, err := c.conn.Recv(-1) if err != nil { c.readError(err) return } if len(buf) > 0 { result, err := c.srv.message.Decode(ctx, buf) if err != nil { glog.Errorf("解析报文失败:%s", err.Error()) } if !c.isReg { c.SetId(ctx.GetId()) c.isReg = true } if ctx.ReportPower { data := new(protocol.PowerData) data.ActivePower = result.ActivePower if err := c.srv.ReportStatus(c.Id, data); err != nil { c.readError(err) return } ctx.SetReportPower(false) } if ctx.ReportVI { data := new(protocol.VIData) data.AV = ctx.GetVIData().AV data.BV = ctx.GetVIData().BV data.CV = ctx.GetVIData().CV data.AI = ctx.GetVIData().AI data.BI = ctx.GetVIData().BI data.CI = ctx.GetVIData().CI if err := c.srv.ReportStatus(c.Id, data); err != nil { c.readError(err) return } ctx.SetReportVI(false) } } } } 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.done) 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(ctx *protocol.PacketContext, powerChan chan struct{}) { defer c.srv.grWG.Done() for { <-powerChan entity := protocol.Dlt_0x33333433{} if ctx.GetReceiveAddress() != nil { sendByte, _ := entity.Encode(ctx) err := c.send(sendByte) if err != nil { glog.Debugf("指令发送失败:%s", err.Error()) } } } } func (c *Client) GetAV(ctx *protocol.PacketContext, avChan, aiChan chan struct{}) { defer c.srv.grWG.Done() for { <-avChan entity := protocol.Dlt_0x33343435{} if ctx.GetReceiveAddress() != nil { sendByte, _ := entity.Encode(ctx) err := c.send(sendByte) if err != nil { glog.Debugf("指令发送失败:%s", err.Error()) } } time.Sleep(1 * time.Second) aiChan <- struct{}{} } } func (c *Client) GetAI(ctx *protocol.PacketContext, aiChan, bvChan chan struct{}) { defer c.srv.grWG.Done() for { <-aiChan entity := protocol.Dlt_0x33343535{} if ctx.GetReceiveAddress() != nil { sendByte, _ := entity.Encode(ctx) err := c.send(sendByte) if err != nil { glog.Debugf("指令发送失败:%s", err.Error()) } } time.Sleep(1 * time.Second) bvChan <- struct{}{} } } func (c *Client) GetBV(ctx *protocol.PacketContext, bvChan, biChan chan struct{}) { defer c.srv.grWG.Done() for { <-bvChan entity := protocol.Dlt_0x33353435{} if ctx.GetReceiveAddress() != nil { sendByte, _ := entity.Encode(ctx) err := c.send(sendByte) if err != nil { glog.Debugf("指令发送失败:%s", err.Error()) } } time.Sleep(1 * time.Second) biChan <- struct{}{} } } func (c *Client) GetBI(ctx *protocol.PacketContext, biChan, cvChan chan struct{}) { defer c.srv.grWG.Done() for { <-biChan entity := protocol.Dlt_0x33353535{} if ctx.GetReceiveAddress() != nil { sendByte, _ := entity.Encode(ctx) err := c.send(sendByte) if err != nil { glog.Debugf("指令发送失败:%s", err.Error()) } } time.Sleep(1 * time.Second) cvChan <- struct{}{} } } func (c *Client) GetCV(ctx *protocol.PacketContext, cvChan, ciChan chan struct{}) { defer c.srv.grWG.Done() for { <-cvChan entity := protocol.Dlt_0x33363435{} if ctx.GetReceiveAddress() != nil { sendByte, _ := entity.Encode(ctx) err := c.send(sendByte) if err != nil { glog.Debugf("指令发送失败:%s", err.Error()) } } time.Sleep(1 * time.Second) ciChan <- struct{}{} } } func (c *Client) GetCI(ctx *protocol.PacketContext, ciChan, powerChan chan struct{}) { defer c.srv.grWG.Done() for { <-ciChan entity := protocol.Dlt_0x33363535{} if ctx.GetReceiveAddress() != nil { sendByte, _ := entity.Encode(ctx) err := c.send(sendByte) if err != nil { glog.Debugf("指令发送失败:%s", err.Error()) } } time.Sleep(time.Duration(g.Cfg().GetInt("Server.PowerFrequency")-5) * time.Second) powerChan <- struct{}{} } } func (c *Client) SendGetAddress(ctx *protocol.PacketContext, addressChan, avChan chan struct{}) { defer c.srv.grWG.Done() <-addressChan bytea := []byte{0x68, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0xAA, 0x68, 0x13, 0x00, 0xDF, 0x16} _ = c.send(bytea) for { time.Sleep(time.Duration(g.Cfg().GetInt("Server.VIFrequency")) * time.Second) avChan <- struct{}{} } }