接收bashboard消息并存储

zhangqian

2023-12-04 3f118683ec4079f6ff4ccca8a8ed080219c1fb8a

接收bashboard消息并存储

 constvar/const.go

@@ -9,6 +9,7 @@
    NsqTopicPullDataRequest           = "aps.%v.pull.data.request"       //拉取数据请求
    NsqTopicPullDataResponse          = "aps.%v.pull.data.response"      //拉取数据响应
    NsqTopicApsClientReportData       = "aps.%v.apsClient.report.data"   //apsClient上报数据
    NsqTopicSendDashboardData         = "aps.%v.apsClient.dashboard"     //下发dashboard数据
)

type PlcStartAddressType int

 nsq/consumer.go

@@ -25,6 +25,8 @@
        handler = &DeviceUpdate{Topic: topic}
    case fmt.Sprintf(constvar.NsqTopicPullDataResponse, conf.Conf.NsqConf.NodeId):
        handler = &PullDataResponse{Topic: topic}
    case fmt.Sprintf(constvar.NsqTopicSendDashboardData, conf.Conf.NsqConf.NodeId):
        handler = &PullDataResponse{Topic: topic}
    }
    c.AddHandler(handler.HandleMessage)
    return

 nsq/msg_handler.go

@@ -433,3 +433,28 @@
    }
    return nil
}

type Dashboard struct {
    Topic string
}

func (slf *Dashboard) HandleMessage(data []byte) (err error) {
    logx.Infof("dashboard message: %s", data)
    var dashboard model.Dashboard
    err = json.Unmarshal(data, &dashboard)
    if err != nil {
        logx.Infof("unmarshal msg err :%s", err)
        return err
    }
    old, err := model.NewDashboardSearch(nil).SetVersion(dashboard.Version).First()
    if err == gorm.ErrRecordNotFound {
        err = model.NewDashboardSearch(nil).SetVersion(dashboard.Version).Create(&dashboard)
    } else {
        err = model.NewDashboardSearch(nil).SetId(old.ID).Updates(&dashboard)
    }
    if err != nil {
        logx.Infof("process pull data  err :%s", err)
        return err
    }
    return nil
}

 nsq/nsq.go

@@ -41,6 +41,7 @@
        constvar.NsqTopicApsProcessParams,
        constvar.NsqTopicDeviceUpdate,
        constvar.NsqTopicPullDataResponse,
        constvar.NsqTopicSendDashboardData,
    }
    for _, t := range topics {
        topic := fmt.Sprintf(t, conf.Conf.NsqConf.NodeId)