完善1400转发流程

zhangzengfei

2024-04-28 62beabfd1466fd3a6b6c0736acc8c5a3b3ec4b3b

完善1400转发流程

 .gitignore

@@ -1,4 +1,4 @@
.idea
gat1400Exchange
gat1400Exchange*
gat1400.db
logs

 Makefile

New file
@@ -0,0 +1,19 @@
VER="3.1.0"
DATA=$(shell date +%Y%m%d)
GIT=$(shell git rev-parse --short HEAD)
BRANCH=$(shell git rev-parse --abbrev-ref HEAD)

build:
    go build -v -a -ldflags "-X main.MinVersion=$(VER) -X main.PublishDate=$(DATA) -X main.CommitId=$(GIT) -X main.Branch=$(BRANCH)"
arm64:
    CGO_ENABLED=1 \
    GOOS=linux \
    GOARCH=arm64 \
    CC=/opt/l4t-gcc/gcc-linaro-7.3.1-2018.05-x86_64_aarch64-linux-gnu/bin/aarch64-linux-gnu-gcc \
    CXX=/opt/l4t-gcc/gcc-linaro-7.3.1-2018.05-x86_64_aarch64-linux-gnu/bin/aarch64-linux-gnu-gcc \
    CGO_CFLAGS="--sysroot=/opt/l4t-gcc/sysroot-glibc-linaro-2.25-2018.05-aarch64-linux-gnu" \
    CGO_CXXFLAGS="--sysroot=/opt/l4t-gcc/sysroot-glibc-linaro-2.25-2018.05-aarch64-linux-gnu" \
    CGO_LDFLAGS="--sysroot=/opt/l4t-gcc/sysroot-glibc-linaro-2.25-2018.05-aarch64-linux-gnu -lstdc++" \
    go build -v -a -tags=agx -ldflags "-X main.MinVersion=$(VER) -X main.PublishDate=$(DATA) -X main.CommitId=$(GIT)" -o gat1400Exchange-arm64
clean:
    rm -f gat1400Exchange*

 client/client.go

@@ -1,15 +1,79 @@
package client

import (
    "context"
    "fmt"
    "math/rand"
    "time"

    "gat1400Exchange/config"
    "gat1400Exchange/pkg/logger"
    "gat1400Exchange/vo"
)

func Init1400Client() {
func Init1400Client(ctx context.Context) {
    if !config.ClientConf.Enable {
        logger.Debug("GAT/1400 Client disabled")
        return
    }

    register()
    go registerLoop(ctx)
    go keepaliveLoop(ctx)
    go syncTimeLoop(ctx)

}

func registerLoop(ctx context.Context) {
    ticker := time.NewTicker(1 * time.Second)
    for {
        select {
        case <-ctx.Done():
            unRegister()
            return
        case <-ticker.C:
            if clientStatus != vo.StatusSuccess {
                fmt.Println("register")
                if !register() {
                    randNum := rand.Intn(300) + 1
                    // 随机等待300s, 重新注册
                    ticker.Reset(time.Duration(randNum) * time.Second)
                } else {
                    ticker.Reset(1 * time.Second)
                }
            }
        }
    }
}

func keepaliveLoop(ctx context.Context) {
    var failCount int
    ticker := time.NewTicker(time.Duration(config.ClientConf.HeartbeatInterval) * time.Second)
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            if status := keepalive(); status != vo.StatusSuccess {
                failCount++

                if failCount > config.ClientConf.HeartbeatFailCount {
                    clientStatus = status
                }
            } else {
                failCount = 0
            }
        }
    }
}

func syncTimeLoop(ctx context.Context) {
    ticker := time.NewTicker(time.Duration(config.ClientConf.HeartbeatInterval*10) * time.Second)
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            syncTime()
        }
    }
}

 client/system.go

@@ -1,11 +1,14 @@
package client

import (
    "encoding/json"
    "fmt"
    "gat1400Exchange/util"
    "io/ioutil"

    "gat1400Exchange/config"
    "gat1400Exchange/pkg/logger"
    "gat1400Exchange/vo"

    dac "github.com/xinsnake/go-http-digest-auth-client"
)
@@ -17,27 +20,90 @@
    TimeUrI       = "/VIID/System/Time"
)

func register() {
    url := fmt.Sprintf("http://%s:%s%s", config.ClientConf.ServerAddr, config.ClientConf.ServerPort, RegisterUrI)
    dr := dac.NewRequest(config.ClientConf.Username, config.ClientConf.Password, "GET", url, "")
var clientStatus = vo.StatusOtherError

var headers = map[string]string{
    "User-Agent":    "AI camera",
    "Accept":        "*",
    "User-Identify": config.ClientConf.ChannelNo,
    "Content-Type":  "application/VIID+JSON",
}

func register() bool {
    url := fmt.Sprintf("%s://%s:%s%s", config.ClientConf.Proto, config.ClientConf.ServerAddr, config.ClientConf.ServerPort, RegisterUrI)
    req := vo.RequestRegister{RegisterObject: vo.RequestDeviceID{DeviceID: config.ClientConf.ChannelNo}}
    reqByte, _ := json.Marshal(req)

    dr := dac.NewRequest(config.ClientConf.Username, config.ClientConf.Password, "POST", url, string(reqByte))
    if headers != nil {
        for k, v := range headers {
            dr.Header.Set(k, v)
        }
        dr.Header.Del("Accept-Encoding")
    }

    resp, err := dr.Execute()
    if err != nil {
        logger.Error(err.Error())
        return
        return false
    }

    defer resp.Body.Close()

    body, err := ioutil.ReadAll(resp.Body)
    if err != nil {
        logger.Error(err.Error())
        return
        return false
    }

    fmt.Printf(string(body))
    fmt.Println("rsp:", string(body))
    var registerResponse vo.ResponseStatus
    err = json.Unmarshal(body, &registerResponse)
    if err != nil {
        fmt.Println("register error,", err.Error())
        return false
    }

    clientStatus = registerResponse.StatusCode

    if registerResponse.StatusCode == vo.StatusSuccess {
        fmt.Println("register success")
        // 注册成功后提交保活和校时
        keepalive()
        syncTime()
    } else {
        fmt.Println("register failure, ", registerResponse.StatusString)
    }

    return clientStatus == vo.StatusSuccess
}

func keepalive() {
func keepalive() int {
    if clientStatus != vo.StatusSuccess {
        return clientStatus
    }

    var body = vo.RequestKeepalive{
        KeepaliveObject: vo.RequestDeviceID{
            DeviceID: config.ClientConf.ChannelNo,
        },
    }

    b, _ := json.Marshal(body)
    rsp, err := util.HttpPost(KeepaliveUrI, headers, b)
    if err != nil {
        logger.Warn("Keepalive request failed, %s", err.Error())
        return vo.StatusOtherError
    }

    var stat vo.ResponseStatus
    err = json.Unmarshal(rsp, &stat)
    if err != nil {
        logger.Warn("Keepalive response unmarshal failed, %s", err.Error())
        return vo.StatusOtherError
    }

    return stat.StatusCode
}

func unRegister() {
@@ -45,5 +111,7 @@
}

func syncTime() {

    if clientStatus != vo.StatusSuccess {
        return
    }
}

 config/config.go

@@ -17,17 +17,17 @@
}

type client struct {
    DeviceID          string `mapstructure:"device-id"`
    Username          string `mapstructure:"username"`
    Password          string `mapstructure:"password"`
    ServerAddr        string `mapstructure:"server-addr"`
    ServerPort        string `mapstructure:"server-port"`
    Proto             string `mapstructure:"proto" `             // http, https
    UploadType        string `mapstructure:"upload-type"`        // binary, url
    ChannelNo         string `mapstructure:"channel-number"`     // 通道号, 同id
    HeartbeatInterval int    `mapstructure:"heartbeat-interval"` // 心跳周期
    HeartbeatCount    int    `mapstructure:"heartbeat-count"`    // 心跳超时次数
    Enable            bool   `mapstructure:"enable"`
    DeviceID           string `mapstructure:"device-id"`
    Username           string `mapstructure:"username"`
    Password           string `mapstructure:"password"`
    ServerAddr         string `mapstructure:"server-addr"`
    ServerPort         string `mapstructure:"server-port"`
    Proto              string `mapstructure:"proto" `             // http, https
    UploadType         string `mapstructure:"upload-type"`        // binary, url
    ChannelNo          string `mapstructure:"channel-number"`     // 通道号, 同id
    HeartbeatInterval  int    `mapstructure:"heartbeat-interval"` // 心跳周期
    HeartbeatFailCount int    `mapstructure:"heartbeat-count"`    // 心跳超时次数
    Enable             bool   `mapstructure:"enable"`
}

type logConfig struct {
@@ -97,5 +97,13 @@
        ForwardConf.RetryInterval = 5
    }

    if ClientConf.HeartbeatInterval == 0 {
        ClientConf.HeartbeatInterval = 30
    }

    if ClientConf.Proto == "" {
        ClientConf.Proto = "http"
    }

    logger.SetLogLevel(LogConf.Level)
}

 main.go

@@ -3,15 +3,14 @@
import (
    "context"
    "fmt"
    "gat1400Exchange/client"
    "gat1400Exchange/cron"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "gat1400Exchange/client"
    "gat1400Exchange/config"
    "gat1400Exchange/cron"
    "gat1400Exchange/models"
    "gat1400Exchange/pkg/logger"
    "gat1400Exchange/routes"
@@ -39,7 +38,8 @@
    }

    // 启动1400客户端
    go client.Init1400Client()
    ctx, cancel := context.WithCancel(context.Background())
    go client.Init1400Client(ctx)

    // 启动网络视频字符叠加器服务
    go service.NVCSServer()
@@ -69,7 +69,7 @@

    // The context is used to inform the server it has 5 seconds to finish
    // the request it is currently handling
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    //ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    if err := srv.Shutdown(ctx); err != nil {
        logger.Error("Server forced to shutdown:", err)

 service/report.go

@@ -71,7 +71,8 @@

        logger.Info("Report device info. %+v", dev)

        _, err = util.HttpPost(config.ForwardConf.ReportServer, nil, data)
        headers := map[string]string{"Content-Type": "applicaiton/json; charset=UTF-8"}
        _, err = util.HttpPost(config.ForwardConf.ReportServer, headers, data)
        if err != nil {
            return err
        }

 util/http.go

@@ -33,7 +33,7 @@
        return nil, err
    }

    req.Header.Set("Content-Type", "applicaiton/json; charset=UTF-8")
    //req.Header.Set("Content-Type", "applicaiton/json; charset=UTF-8")
    if header != nil {
        for k, v := range header {
            req.Header.Set(k, v)

 vo/message.go

@@ -8,6 +8,10 @@
    KeepaliveObject RequestDeviceID `json:"KeepaliveObject"`
}

type RequestRegister struct {
    RegisterObject RequestDeviceID `json:"RegisterObject"`
}

type RequestUnRegister struct {
    UnRegisterObject RequestDeviceID `json:"UnRegisterObject"`
}