|
|
@@ -1,9 +1,43 @@
|
|
|
package reporter
|
|
|
|
|
|
-import "hnyfkj.com.cn/rtu/linux/baseapp"
|
|
|
+import (
|
|
|
+ "context"
|
|
|
+ "io/fs"
|
|
|
+ "os"
|
|
|
+ "path/filepath"
|
|
|
+ "sync/atomic"
|
|
|
+
|
|
|
+ mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
+ mcu "hnyfkj.com.cn/rtu/xy_v/mcu_ctrl_board"
|
|
|
+
|
|
|
+ "hnyfkj.com.cn/rtu/linux/baseapp"
|
|
|
+ "hnyfkj.com.cn/rtu/linux/netmgrd"
|
|
|
+ "hnyfkj.com.cn/rtu/linux/utils/ftpclient"
|
|
|
+ "hnyfkj.com.cn/rtu/linux/utils/singletask"
|
|
|
+)
|
|
|
|
|
|
const MODULE_NAME = "Reporter"
|
|
|
|
|
|
+var (
|
|
|
+ Reporter *MQTTReporter
|
|
|
+)
|
|
|
+
|
|
|
+type MQTTReporter struct {
|
|
|
+ client mqtt.Client
|
|
|
+
|
|
|
+ ctx context.Context
|
|
|
+ cancel context.CancelFunc
|
|
|
+
|
|
|
+ isLogin atomic.Bool // 标记是否已成功登录MQTT后端服务器
|
|
|
+ dui string // 登录成功后服务端返回的设备唯一ID
|
|
|
+ inheritDUI string // 继承的历史ID, 可选可为空(换板时)
|
|
|
+
|
|
|
+ // 主动上报的后台任务, 登录成功时用于照片补录和上报通知类消息
|
|
|
+ uploadPhotosTask *singletask.OnceTask // 补录照片, 单实例
|
|
|
+ reportMcuCfgTask *singletask.OnceTask // 上报配置, 单实例
|
|
|
+ reportRtuPosTask *singletask.OnceTask // 上报位置, 单实例
|
|
|
+}
|
|
|
+
|
|
|
func ModuleInit() bool {
|
|
|
err := loadCfgServers()
|
|
|
if err != nil {
|
|
|
@@ -13,3 +47,62 @@ func ModuleInit() bool {
|
|
|
|
|
|
return true
|
|
|
}
|
|
|
+
|
|
|
+func (r *MQTTReporter) IsLogin() bool {
|
|
|
+ return r.isLogin.Load()
|
|
|
+}
|
|
|
+
|
|
|
+func (r *MQTTReporter) OnLogin(client mqtt.Client, msg mqtt.Message) {
|
|
|
+ imei, dui, err := parseLoginResp(msg.Payload())
|
|
|
+ if err != nil {
|
|
|
+ baseapp.Logger.Errorf("[%s] 登录失败: %v!!", MODULE_NAME, err)
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ if imei != netmgrd.GetIMEI() { // 判断是否我的应答
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ if len(r.inheritDUI) > 0 && r.inheritDUI == dui {
|
|
|
+ _ = os.Remove(filepath.Join(baseapp.VAR_DIR, "inheritDUI.txt"))
|
|
|
+ } else if len(r.inheritDUI) > 0 && r.inheritDUI != dui {
|
|
|
+ baseapp.Logger.Errorf("[%s] 登录失败: 要继承的DUI(%s)与服务器返回的DUI(%s)不匹配!!", MODULE_NAME, r.inheritDUI, dui)
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ r.dui = dui
|
|
|
+ r.isLogin.Store(true)
|
|
|
+ r.client.Unsubscribe("/yfkj/bxs-sy/server/rpc/response") // 登录成功后, 取消对登录应答的订阅(不关心是否取消成功)
|
|
|
+ baseapp.Logger.Infof("[%s] 登录成功, 设备DUI: %s", MODULE_NAME, r.dui)
|
|
|
+
|
|
|
+ r.uploadPhotosTask.Run(r.uploadPendingPhotos, true) // 补录上传历史遗存照片, 单实例运行
|
|
|
+}
|
|
|
+
|
|
|
+func (r *MQTTReporter) uploadPendingPhotos() {
|
|
|
+ baseapp.Logger.Infof("[%s] 拍照补录上传任务开始", MODULE_NAME)
|
|
|
+ nums := 0
|
|
|
+ filepath.WalkDir(baseapp.IMG_DIR, func(path string, d fs.DirEntry, walkErr error) error {
|
|
|
+ if walkErr != nil || d.IsDir() || filepath.Ext(path) != ".jpg" {
|
|
|
+ return nil
|
|
|
+ } else if !r.IsLogin() || r.ctx.Err() != nil {
|
|
|
+ return context.Canceled
|
|
|
+ }
|
|
|
+
|
|
|
+ ftpclient.FileUploader.Lock() // 上传锁定(多个上传任务并发时, 保证串行执行)
|
|
|
+ defer ftpclient.FileUploader.Unlock()
|
|
|
+
|
|
|
+ mcu.GlobalWorkState.Add(mcu.PhotoUploading)
|
|
|
+ defer mcu.GlobalWorkState.Remove(mcu.PhotoUploading)
|
|
|
+
|
|
|
+ _, err := ftpclient.UploadFileToFtp(r.ctx, path, CfgServers.Img2Ftp.Address, CfgServers.Img2Ftp.Username, CfgServers.Img2Ftp.Password, ftpclient.DefaultUploadTimeout)
|
|
|
+ if err == nil {
|
|
|
+ baseapp.Logger.Infof("[%s] 拍照补录上传成功, 本地文件: %q已删除", MODULE_NAME, path)
|
|
|
+ os.Remove(path)
|
|
|
+ nums++
|
|
|
+ }
|
|
|
+
|
|
|
+ return nil
|
|
|
+ })
|
|
|
+
|
|
|
+ baseapp.Logger.Infof("[%s] 拍照补录上传任务结束, 本次共上传%d张照片", MODULE_NAME, nums)
|
|
|
+}
|
niujiuru