dlt645-server/server/client.go

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
	gatewayId     uint16
	isReg         bool
}

func NewClient(s *Server, conn *gtcp.Conn) *Client {
	return &Client{
		srv:       s,
		conn:      conn,
		sendChan:  make(chan []byte),
		closeChan: make(chan struct{}),
	}
}

func (c *Client) ReadLoop(ctx *protocol.PacketContext, data *protocol.Data) {
	defer c.srv.grWG.Done()
	for {
		buf, err := c.conn.Recv(-1)
		if err != nil {
			c.readError(err)
			return
		}
		if len(buf) > 0 {
			_, err := c.srv.message.Decode(ctx, buf, data)
			if err != nil {
				glog.Errorf("解析报文失败:%s", err.Error())
			}
			if !c.isReg {
				c.SetId(ctx.GetId())
				c.isReg = true
			}
		}
	}
}

func (c *Client) ReportPower(data *protocol.Data) {
	defer c.srv.grWG.Done()
	for {
		reportData := new(protocol.PowerData)
		reportData.ActivePower = data.ActivePower
		if err := c.srv.ReportStatus(c.Id, reportData); err != nil {
			c.readError(err)
			return
		}
		time.Sleep(time.Duration(g.Cfg().GetInt("Server.PowerFrequency")) * time.Second)
	}
}

func (c *Client) ReportVIData(data *protocol.Data) {
	defer c.srv.grWG.Done()
	for {
		reportData := new(protocol.VIData)
		reportData.AV = data.AV
		reportData.BV = data.BV
		reportData.CV = data.CV
		reportData.AI = data.AI
		reportData.BI = data.BI
		reportData.CI = data.CI
		if err := c.srv.ReportStatus(c.Id, reportData); err != nil {
			c.readError(err)
			return
		}
		time.Sleep(time.Duration(g.Cfg().GetInt("Server.VIFrequency")) * time.Second)
	}

}

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(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(2 * 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(2 * 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(2 * 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(2 * 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(2 * 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{}{}
	}
}