testbench/tcp/bds/bdsConn.go

package bds

import (
    "testbench/tcp/tcpserver"
    "fmt"
    "github.com/astaxie/beego"
    "net"
    "time"
    "wb/ut"
    "testbench/models/statusMgr"
    "errors"
    "testbench/tcp/tc"
    "runtime/debug"
    "strings"
    "wb/lg"
    "wb/modbus"
    "reflect"
)

type bdsConn struct {
    tc.TConn
    msgBuff   []byte
    //TermIdByte []byte
    sendSn    uint16
    IsLogin   bool
    msgStart  int
    msgEnd    int
    Uid       string
    stopChan  chan int
    ModelInfo modbus.ModelInfo
}

func newBdsConn(conn net.Conn) *bdsConn {
    o := bdsConn{}
    o.Conn = conn
    o.Typo = "BDS"
    o.sendSn = 0
    o.IsLogin = false
    o.IsConnect = true
    o.msgBuff = make([]byte, 4096)
    o.msgStart = 0
    o.msgEnd = 0
    o.stopChan = make(chan int, 16)
    return &o
}

func (this *bdsConn)DoRecvRegister(req []byte) bool {
    this.LogInfo("REGISTER")
    body := make([]byte, 19)
    timeBytes := time.Now().Format("20060102150405")
    body[0] = req[msgSnStart]
    body[1] = req[msgSnStart + 1]
    body[2] = 0
    body[3] = 0x54
    body[4] = 0x52
    for i, v := range []byte(timeBytes) {
        body[i + 5] = v
    }
    this.sendMsg(mtSvcTermRegister, body)
    return true
}
func (this *bdsConn)DoRecvAuth(req []byte) bool {
    this.LogInfo("AUTH")
    termId := req[msgBodyStart:len(req) - 2]
    if !this.SetTermId(termId){
        return false
    }
    if this.TermId != "" {
        this.InitLog(this.TermId)
        tc.AddMConn(this)
    } else {
        lg.Error("bds EchoFunc TermId not allowed")
    }
    this.IsLogin = true
    this.sendCommonResponseOk(req)
    return true
}
func (this *bdsConn)DoRecvPosition(req []byte) {
    this.LogInfo("POS UPLOAD", ut.BytesToHexStr(req))
    this.sendCommonResponseOk(req)
    if this.TermId == ""{
        return
    }
    x, y, ok := getPosition(req)
    if ok{
        this.AddPosition(this.TermId, x, y)
    }
    return
}
func (this *bdsConn)DoRecvHeartBeat(req []byte) bool {
    this.LogInfo("HTBT")
    this.sendCommonResponseOk(req)
    this.RefreshStatus()
    return true
}
func (this *bdsConn)DoRecvUart(recv []byte) bool {
    if this.TermId == ""{
        return false
    }
    if len(recv) < msgUartMinLen {
        this.LogError("[U] Recive msg len error:", ut.BytesToHexStr(recv))
        return true
    }
    switch recv[msgBodyStart + 2] {
    case 0x03, 0x02, 0x01:
        mdResp := recv[mdbsStart:len(recv) - 2]
        modbus.DoRecvMdbsUart(this.ModelInfo, this.StatusMap, mdResp, this.Logger)
        this.StatusMap["status"] = "online"
        if rpm, ok :=ut.Maps.GetFloat64(this.StatusMap, "rpm");ok{
            if rpm > 0{
                this.StatusMap["status"] = "running"
            }
        }else{
            this.LogError("rpm is not float64!", reflect.TypeOf(rpm))
        }
        if this.ModelInfo.CreateAlarm(this.StatusMap){
            this.StatusMap["status"] = "alarm"
        }
        this.AddStatus(ut.Maps.Copy(this.StatusMap))
    case 0x10:
        this.LogInfo("[U][CMD]Recive uart cmd response")
    default:
        this.LogError("[U][OTHER]Recive uart other response")
    }
    return true
}
func (this *bdsConn) SendCmd(key string) {
    this.LogInfo("[CMD]:", key)
    switch key {
    case "deletesample":
        this.sendClearSampleCmd()
        return
    case "initsample":
        this.sendInitSampleCmd(this.ModelInfo)
        return
    }
    var req []byte
    if cmd, ok := this.ModelInfo.CmdMap[key];ok{
        req = cmd.Bytes
        if cmd.Type == modbus.MsgTypeIo{
            this.sendMsg(mtSvcIo, req)
        }else{
            req = modbus.AppendCrc(req)
            body := ut.BsPrepend(req, uartBit)
            this.sendMsg(mtSvcQueryUart, body)
        }
        this.LogInfo("MConn.SendCmd req:",req)
    }else{
        this.LogError("MConn.SendCmd error no such cmd:",key)
    }
}
func (this *bdsConn) sendClearSampleCmd(){
    for i:= 1 ; i< 10 ; i++{
        body := []byte{0x41, 0x00}
        body = append(body, byte(i))
        this.sendMsg(msSvcDeleteQueryUart, body)
        time.Sleep(time.Second)
    }
}
func (this *bdsConn) sendInitSampleCmd(mdif modbus.ModelInfo){
    timeConifg := []byte{0x41, 0x00, 0x00, 0x28, 0x00}
    this.sendMsg(msSvcInitQueryUartSampleTime, timeConifg)
    time.Sleep(2*time.Second)
    i := int8(1)
    for _, query := range mdif.Querys{
        req := modbus.BuildReadRequest(query.Address, query.Code, query.RegStart, query.RegLen)
        body := []byte{0x41, 0x00}
        body = append(body, byte(i))
        i = i + 1
        body = append(body, req...)
        this.sendMsg(msSvcInitQueryUart, body)
        time.Sleep(2*time.Second)
    }
}
func (this *bdsConn) InitDtu(startIds ...string)bool{
    if len(startIds) > 0{
        startId := startIds[0]
        if mdif, ok := modbus.ModelInfoMap[startId];ok{
            //this.sendClearSampleCmd()
            //time.Sleep(5*time.Second)
            this.sendInitSampleCmd(mdif)
        }else{
            lg.Error("bdsConn.SendInitSmapleCmd: no such startId ", startId)
            return false
        }
    }else{
        this.sendInitSampleCmd(this.ModelInfo)
    }
    return true
}
func (this *bdsConn)DoTermCommonResp(req []byte) bool {
    return true
}

func (this *bdsConn) SetTermId(termIdByte []byte)bool {
    this.TermId = string(termIdByte)
    mStartId := ""
    for startId, modelInfo := range modbus.ModelInfoMap{
        if strings.HasPrefix(this.TermId, startId){
            if len(mStartId) > len(startId){
                continue
            }
            this.ModelInfo = modelInfo
            mStartId = startId
        }
    }
    if mStartId == ""{
        this.LogError("bdsConn.SetTermId modelInfo error: no model match id:", this.TermId)
        return false
    }
    this.LogInfo("MConn.Init modelInfo as: ", this.ModelInfo.Name)
    lg.Info("bdsConn.SetTermId: init statusMgr id:", this.TermId)
    this.StatusMgr = statusMgr.GetMgrBySid(this.TermId)
    this.StatusMap = this.StatusMgr.GetDefaultStatusMap(this.TermId)
    return true
}
func (this *bdsConn)sendMsg(mt, msgBody []byte) {
    msgBodyLen := len(msgBody)
    msgLen := msgBodyLen + msgMinLen
    resp := make([]byte, msgLen)
    resp[0] = startBit
    resp[msgTypeStart] = mt[0]
    resp[msgTypeStart + 1] = mt[1]
    resp[msgHaedStart] = byte((msgBodyLen >> 8) & 0x01)
    resp[msgHaedStart + 1] = byte(msgBodyLen)
    resp[msgSnStart] = byte(this.sendSn >> 8)
    resp[msgSnStart + 1] = byte(this.sendSn)
    this.sendSn = this.sendSn + 1
    for i, b := range msgBody {
        resp[msgBodyStart + i] = b
    }
    resp[msgBodyStart + len(msgBody)] = getCheckSum(resp)
    resp[msgBodyStart + len(msgBody) + 1] = stopBit
    msg := EscapeChars(resp)
    this.LogDebug("[S]: ", msg)
    this.LogInfo("[S][P]: ", parseMsg(resp))
    if _, err:= this.Write(msg); err != nil{
        this.IsConnect = false
        this.LogWarn("[E]: sendMsg error :", err.Error())
        return
    }
}
func (this *bdsConn)isAvaliableMsg(req []byte) bool {
    if len(req) < msgMinLen {
        this.LogError(fmt.Sprintf("RECEIVE TERMINAL[%s]:", this.TermId), "MESSAGE FORMAT LEN ERROR")
        return false
    }
    if req[0] != 0x7E {
        this.LogError(fmt.Sprintf("RECEIVE TERMINAL[%s]:", this.TermId), "MESSAGE FORMAT START EEROR")
        return false
    }
    if req[len(req) - 1] != 0x7E {
        this.LogError(fmt.Sprintf("RECEIVE TERMINAL[%s]:", this.TermId), "MESSAGE FORMAT END EEROR")
        return false
    }
    return true
}
// 发送平台通用应答
func (this *bdsConn)sendCommonResponse(req, ret []byte) {
    body := make([]byte, len(ret) + 4)
    body[0] = req[msgSnStart]
    body[1] = req[msgSnStart + 1]
    body[2] = req[msgTypeStart]
    body[3] = req[msgTypeStart + 1]
    for i, v := range ret {
        body[i + 4] = v
    }
    this.sendMsg(mtSvcCommonResponse, body)
}
// 发送平台通用应答成功
func (this *bdsConn)sendCommonResponseOk(req []byte) {
    this.sendCommonResponse(req, []byte{0x00})
}
const (
    statInit = 0
    statStart = 1
    statContent = 2
)
func (this *bdsConn) Write(req []byte) (n int, err error) {
    n, err = this.Conn.Write(req)
    this.LogSend(ut.BytesToHexStr(req))
    return n, err
}
func (this *bdsConn)getMsg() ([]byte, error) {
    buf := make([]byte, 4096)
    retMsg := make([]byte, 0, 256)
    // 读取次数，如果超过255次还不能读到一个完整的msg则重新连接
    readNum := 0
    status := statInit
    //this.LogDebug("[M]getMsg start:", " msgStart: ", this.msgStart, " msgEnd:", this.msgEnd)
    //if this.msgEnd > 0 {
        //this.LogDebug("[M]getMsg start:", " msgBuf: ", this.msgBuff[:this.msgEnd])
    //}
    fillMsg:
    // 排除越界问题
    if this.msgStart >= 1024 || this.msgEnd > 1024 || this.msgStart >= this.msgEnd || this.msgEnd <= 0 {
        //this.LogDebug("[M]fillMsg start need read new: read num ", readNum, " msgStart: ", this.msgStart, " msgEnd:", this.msgEnd)
        this.msgStart = 0
        this.msgEnd = 0
    } else {
        //this.LogDebug("[M]fillMsg start: read num ", readNum, " msgStart: ", this.msgStart, " msgEnd:", this.msgEnd)
        //this.LogDebug("[M]fillMsg start msgBuf:", ut.BytesToHexStr(this.msgBuff[:this.msgEnd]))
        if len(retMsg) > 0 {
            this.LogDebug("[M]fillMsg start ret msg:", retMsg)
        }
        for i := this.msgStart; i < this.msgEnd; i++ {
            this.msgStart = i + 1
            switch status {
            case statInit:
                if this.msgBuff[i] == startStopBit {
                    retMsg = append(retMsg, startStopBit)
                    status = statStart
                    //this.LogDebug("[M]statInit: msg start")
                    continue
                } else {
                    this.LogWarn("[M]statInit: Read msg start error got:", this.msgBuff[i])
                }
            case statStart:
                // 连续两个开始，忽略前面一个开始
                if this.msgBuff[i] == startStopBit {
                    this.LogWarn("[M]statStart: mulite start bit")
                } else {
                    retMsg = append(retMsg, this.msgBuff[i])
                    status = statContent
                }
            case statContent:
                retMsg = append(retMsg, this.msgBuff[i])
                if this.msgBuff[i] == startStopBit {
                    if this.msgStart == this.msgEnd {
                        this.msgStart = 0
                        this.msgEnd = 0
                    }
                    return retMsg, nil
                }
            }
        }
    }
    if readNum < 255 {
        //this.LogInfo("START READ REAL")
        i, err := this.Conn.Read(buf)
        this.msgStart = 0
        if err != nil {
            this.LogWarn("[M]Read bytes from [", this.TermId, "] error", err.Error())
            this.msgStart = 0
            this.msgEnd = 0
            return nil, err
        }
        //this.LogInfo("START READ REAL END")
        this.msgEnd = i
        readNum = readNum + 1
        this.msgBuff = buf[:this.msgEnd]
        this.LogRecv(ut.BytesToHexStr(buf[:this.msgEnd]))
        goto fillMsg
    }
    this.LogWarn("[M]Error receive re from:", this.TermId, " Read 255 time get no msg")
    this.msgStart = 0
    this.msgEnd = 0
    return nil, errors.New("No msg error")
}

func EchoFunc(conn net.Conn) {
    lg.Info("BDS CONNECT FROM: ", conn.RemoteAddr())
    defer conn.Close()
    bConn := newBdsConn(conn)
    defer func() {
        bConn.Close()
        bConn.stopChan <- 1
        if bConn.TermId != "" {
            tc.DeleteConn(bConn)
        }
        if err := recover(); err != nil{
            lg.Error("EchoFunc panic", err)
            bConn.LogError("[T] EchoFunc panic traceback:", string(debug.Stack()))
        }
    }()
    isTestSendStart := false
    for bConn.IsConnect {
        if bConn.IsLogin && (isTestSendStart == false) {
            if needSendTest == true{
                go TestSend(bConn)
            }
            isTestSendStart = true
        }
        msg, err := bConn.getMsg()
        if err != nil {
            lg.Error("Read msg from ", bConn.TermId, " error: ", err.Error())
            bConn.IsConnect = false
            return
        }
        if !bConn.isAvaliableMsg(msg) {
            bConn.LogWarn("BDS RECV MSG ERROR: ", ut.BytesToHexStr(msg))
            continue
        }
        //bConn.LogInfo("loop ---------------", bConn.Uid)
        bConn.LogDebug("[R][P]: ", parseMsg(msg))
        req := RestoreEscChars(msg)
        switch req[msgTypeStart] {
        // 注册类
        case 0x01:
            switch req[msgTypeStart + 1] {
            // 终端注册
            case 0x00:
                bConn.DoRecvRegister(req)
                continue
            // 终端认证
            case 0x02:
                bConn.DoRecvAuth(req)
                continue
            }
        // 通用类
        case 0x00:
            switch req[msgTypeStart + 1] {
            // 终端通用应答
            case 0x01:
                bConn.DoTermCommonResp(req)
                continue
            //终端心跳
            case 0x02:
                bConn.DoRecvHeartBeat(req)
                continue
            }
        case 0x02:
            switch req[msgTypeStart + 1] {
            case 0x00:
                bConn.DoRecvPosition(req)
                continue
            }
        // 串口消息
        case 0x09:
            switch req[msgTypeStart + 1] {
            case 0x00:
                bConn.DoRecvUart(req)
                continue
            }
        }
        bConn.sendCommonResponseOk(req)
    }
}

// 几秒后收集
//func CollectAfter(sn string, waitLen int64) {
//    lg.Info("CollectAfter:", waitLen)
//}
func TestSend(conn *bdsConn){
    conn.LogInfo("TestSend start")
    var i uint64
    i = 0
    for conn.IsConnect{
        time.Sleep(2*time.Second)
        i += 1
        //fmt.Println("send test")
        req := []byte(fmt.Sprintf("-----this is test %d=====\r\n", i))
        body := ut.BsPrepend(req, uartBit)
        conn.sendMsg(mtSvcQueryUart, body)
    }
    conn.LogInfo("TestSend end")
}
func ServerRun() {
    port := beego.AppConfig.String("bdsport")
    server.Run(port, EchoFunc)
}
var queryUartDelay time.Duration
var needSendTest bool
func init(){
    samp := beego.AppConfig.DefaultInt("samplingPeriods", 20)
    queryUartDelay = time.Duration(samp)
    lg.Info("Bds samplingPeriods:", queryUartDelay)
    needSendTest = beego.AppConfig.DefaultBool("needSendTest", false)
    lg.Info("Bds needSendTest:", needSendTest)
}