bth-rs30-gateway/server/client.go

package server

import (
	"bth-rs30-gateway/protocol"
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"github.com/gogf/gf/encoding/gbinary"
	"github.com/gogf/gf/net/gtcp"
	"github.com/gogf/gf/os/glog"
	"github.com/gogf/gf/util/gconv"
	"io"
	"math"
	"net"
	"strconv"
	"strings"
	"syscall"
	"time"
)

type Client struct {
	Id           string
	srv          *Server
	conn         *gtcp.Conn
	sendChan     chan []byte
	closeChan    chan struct{}
	closeHandler func(id string, c *Client)
	regHandler   func(id string, c *Client)
	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) SetId(id string) {
	c.Id = id
}

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.Error("接收指令超时")
					break
				default:
					receiveBuf, err := c.conn.Recv(-1)
					if err != nil {
						c.readError(err)
						break
					}
					if !c.isReg {
						id := gbinary.DecodeToString(receiveBuf)
						glog.Debugf("收到注册包!id:%s", id)
						c.SetId(id)
						c.isReg = true
						if c.regHandler != nil {
							c.regHandler(c.Id, c)
						}
						continue
					}
					glog.Debugf("收到数据：%2X", receiveBuf)
					if err := c.decodeAndReport(receiveBuf); err != nil {
						glog.Debugf("处理数据失败:%s", err.Error())
						break
					}
				}
				break
			}
		}
	}
}

// 收到数据：01 03 04 00 FD 00 8A EA 64
func (c *Client) decodeAndReport(buf []byte) error {
	length := len(buf)
	var crc crc
	crc.reset().pushBytes(buf[0 : length-2])
	checksum := uint16(buf[length-1])<<8 | uint16(buf[length-2])
	if checksum != crc.value() {
		return errors.New(fmt.Sprintf("modbus: response crc '%v' does not match expected '%v'", checksum, crc.value()))
	}

	if buf[1] == 0x03 {
		if buf[2] == 0x04 {
			data := &protocol.Data{}
			data.Temperature = gconv.Float32(caleTemperature(buf[3:5])) / 10
			data.Humidly = gconv.Float32(gbinary.BeDecodeToUint16(buf[5:7])) / 10
			if err := c.srv.ReportStatus(c.Id, data, "status"); err != nil {
				return err
			}
		}
		if buf[2] == 0x08 {
			data := &protocol.OffsetData{
				TemOffset: 2,
				HumOffset: 2,
			}
			if gconv.Int(gbinary.BeDecodeToUint16(buf[3:5])) == 1 {
				data.TemOffset = 1
				value := gconv.Float32(gbinary.BeDecodeToUint16(buf[5:7])&0x7FFF) / 100
				if gconv.Int(gbinary.BeDecodeToUint16(buf[5:7])&0x8000) == 0 {
					value = -value
				}
				data.TemOffsetValue = value
			}
			if gconv.Int(gbinary.BeDecodeToUint16(buf[7:9])) == 1 {
				data.HumOffset = 1
				data.HumOffsetValue = gconv.Float32(gbinary.BeDecodeToUint16(buf[9:11])) / 100
			}
			if err := c.srv.ReportStatus(c.Id, data, "rs30_offset"); err != nil {
				return err
			}
		}
	}
	return nil
}
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)
	}
}

// 计算温度值, 处理零下的情况
func caleTemperature(data []byte) int {
	var ym uint16
	var isBlowZero bool
	var result int
	bm := binary.BigEndian.Uint16(data)
	var bitNum = len(data) * 8
	f := "%." + strconv.Itoa(bitNum) + "b"
	bmStr := fmt.Sprintf(f, bm)
	if string(bmStr[0]) == "1" { // blow zero
		ym = ^bm + 1
		isBlowZero = true
	} else {
		ym = bm
	}
	result = int(ym)
	if isBlowZero {
		result = int(ym) * -1
	}
	return result
}

// 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
}

// SetOffset 设置温湿度偏移值
func (c *Client) SetOffset(act int, value float32, slaveId int) error {
	var sendBuf []byte
	buffer := bytes.NewBuffer(sendBuf)
	buffer.Write(gbinary.BeEncodeInt(slaveId))

	var funcByte byte
	var address []byte
	if act == 1 || act == 3 {
		funcByte = 0x10
	} else if act == 2 || act == 4 {
		funcByte = 0x06
	}
	if act == 1 || act == 2 {
		address = []byte{0x00, 0x50}
	} else if act == 3 || act == 4 {
		address = []byte{0x00, 0x52}
	}
	buffer.WriteByte(funcByte)
	buffer.Write(address)

	if act == 1 {
		number := []byte{0x00, 0x02}
		buffer.Write(number)
		buffer.Write([]byte{0x00, 0x01})
		buffer.Write(dataBlock(value))
	} else if act == 3 {
		number := []byte{0x00, 0x02}
		buffer.Write(number)
		buffer.Write([]byte{0x00, 0x01})
		buffer.Write(gbinary.BeEncodeUint16(uint16(value * 100)))
	} else if act == 2 || act == 4 {
		buffer.Write([]byte{0x00, 0x00})
	}
	var mCrc crc
	checkSum := mCrc.reset().pushBytes(buffer.Bytes()).value()
	buffer.Write([]byte{byte(checkSum), byte(checkSum >> 8)})
	//fmt.Printf("%2X", buffer.Bytes())
	c.sendChan <- buffer.Bytes()
	return nil
}

func dataBlock(value float32) []byte {
	buffer := &bytes.Buffer{}
	var a uint16
	if value >= 0 {
		a = uint16(value*100) | 0x8000
	} else {
		a = uint16(math.Abs(float64(value))*100) | 0x0000
	}
	buffer.Write(gbinary.BeEncodeUint16(a))
	return buffer.Bytes()
}

func (c *Client) GetOffset() {
	c.sendChan <- []byte{0x01, 0x03, 0x00, 0x50, 0x00, 0x04, 0x44, 0x18}
}

func (c *Client) GetSendByte() {
	for {
		c.sendChan <- []byte{0x01, 0x03, 0x00, 0x00, 0x00, 0x02, 0xC4, 0x0B}
		time.Sleep(10 * time.Second)
	}
}

func (c *Client) send(buf []byte) error {
	if c.conn == nil {
		return nil
	}
	glog.Debugf("----->%2X", buf)
	err := c.conn.Send(buf)
	if err != nil {
		glog.Error(c.srv.ctx, err)
		c.closeConnection()
		return err
	}
	return nil
}