dlt645-server/server/client.go

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 <-c.closeChan:
					return
				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.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 <-c.closeChan:
			return nil
		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
					continue
				}
				_, err = c.srv.message.Decode(c.packetContext, receiveBuf)
				if err != nil {
					glog.Errorf("解析报文失败:%s", err.Error())
					return err
				}
				c.SetId(c.packetContext.GetDeviceId())
				if c.regHandler != nil {
					c.regHandler(c.Id, c)
				}
				return nil
			}
		}
	}
}