allied-creation / allied-creation-manufacture · 提交

转到一个项目

作者 yangfu
提交 6a5c5a44a219991955813046a2348839b4c5fe1e 1 个父辈 e0d29926

feat:1.MQTT数据汇报处理

内嵌 并排对比
正在显示 8 个修改的文件 包含 332 行增加 和 188 行删除
@@ -15,7 +15,7 @@ type CreateDeviceCollectionCommand struct { @@ -15,7 +15,7 @@ type CreateDeviceCollectionCommand struct {
15 // 车间名 15 // 车间名
16 WorkShopName string `cname:"车间名" json:"workShopName" valid:"Required"` 16 WorkShopName string `cname:"车间名" json:"workShopName" valid:"Required"`
17 // 启动状态 1-启动 0-停止 17 // 启动状态 1-启动 0-停止
18 - StartupStatus int64 `cname:"启动状态 1-启动 0-停止" json:"startupStatus"` 18 + StartupStatus int64 `cname:"启动状态 1-启动 0-停止" json:"startupStatus"`
19 // 采集时间 19 // 采集时间
20 CollectionTime time.Time `cname:"采集时间" json:"collectionTime" valid:"Required"` 20 CollectionTime time.Time `cname:"采集时间" json:"collectionTime" valid:"Required"`
21 // 设备名 21 // 设备名
@@ -26,6 +26,8 @@ type CreateDeviceCollectionCommand struct { @@ -26,6 +26,8 @@ type CreateDeviceCollectionCommand struct {
26 ComStatus int64 `cname:"通讯状态 1-通讯正常 0-设备未上电或与采集端通讯故障" json:"comStatus"` 26 ComStatus int64 `cname:"通讯状态 1-通讯正常 0-设备未上电或与采集端通讯故障" json:"comStatus"`
27 // 设备数据值 27 // 设备数据值
28 Values map[string]interface{} `cname:"设备数据值" json:"values" valid:"Required"` 28 Values map[string]interface{} `cname:"设备数据值" json:"values" valid:"Required"`
  29 + // 预查
  30 + PreCheck bool `json:"check"`
29 } 31 }
30 32
31 func (createDeviceCollectionCommand *CreateDeviceCollectionCommand) Valid(validation *validation.Validation) { 33 func (createDeviceCollectionCommand *CreateDeviceCollectionCommand) Valid(validation *validation.Validation) {
1 package service 1 package service
2 2
3 import ( 3 import (
  4 + "errors"
4 "fmt" 5 "fmt"
5 "github.com/linmadan/egglib-go/core/application" 6 "github.com/linmadan/egglib-go/core/application"
6 "github.com/linmadan/egglib-go/utils/tool_funs" 7 "github.com/linmadan/egglib-go/utils/tool_funs"
@@ -8,26 +9,27 @@ import ( @@ -8,26 +9,27 @@ import (
8 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/deviceCollection/query" 9 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/deviceCollection/query"
9 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/factory" 10 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/application/factory"
10 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain" 11 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
  12 + "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/domainService"
  13 + "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/redis"
  14 + "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
  15 + "strconv"
  16 + "sync"
  17 + "time"
11 ) 18 )
12 19
13 type DeviceCollectionService struct { 20 type DeviceCollectionService struct {
14 } 21 }
15 22
  23 +var (
  24 + DeviceDataCache = NewDeviceDataInstance()
  25 +)
  26 +
16 // 创建 27 // 创建
17 func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(createDeviceCollectionCommand *command.CreateDeviceCollectionCommand) (interface{}, error) { 28 func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(createDeviceCollectionCommand *command.CreateDeviceCollectionCommand) (interface{}, error) {
18 if err := createDeviceCollectionCommand.ValidateCommand(); err != nil { 29 if err := createDeviceCollectionCommand.ValidateCommand(); err != nil {
19 return nil, application.ThrowError(application.ARG_ERROR, err.Error()) 30 return nil, application.ThrowError(application.ARG_ERROR, err.Error())
20 } 31 }
21 - transactionContext, err := factory.CreateTransactionContext(nil)  
22 - if err != nil {  
23 - return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())  
24 - }  
25 - if err := transactionContext.StartTransaction(); err != nil {  
26 - return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())  
27 - }  
28 - defer func() {  
29 - transactionContext.RollbackTransaction()  
30 - }() 32 +
31 newDeviceCollection := &domain.DeviceCollection{ 33 newDeviceCollection := &domain.DeviceCollection{
32 //DeviceCollectionId: createDeviceCollectionCommand.DeviceCollectionId, 34 //DeviceCollectionId: createDeviceCollectionCommand.DeviceCollectionId,
33 WorkShopName: createDeviceCollectionCommand.WorkShopName, 35 WorkShopName: createDeviceCollectionCommand.WorkShopName,
@@ -38,6 +40,31 @@ func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(c @@ -38,6 +40,31 @@ func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(c
38 CollectionTime: createDeviceCollectionCommand.CollectionTime, 40 CollectionTime: createDeviceCollectionCommand.CollectionTime,
39 Values: createDeviceCollectionCommand.Values, 41 Values: createDeviceCollectionCommand.Values,
40 } 42 }
  43 + var lastDeviceCollectionRecord = &domain.DeviceCollection{}
  44 + var err error
  45 + if createDeviceCollectionCommand.PreCheck {
  46 + //前置验证,限制设备上报速率
  47 + if lastDeviceCollectionRecord, err = DeviceDataCache.Add(newDeviceCollection.DeviceSn, newDeviceCollection.DeviceType, newDeviceCollection); err != nil {
  48 + //log.Logger.Error(err.Error()+newDeviceCollection.DeviceSn)
  49 + return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  50 + }
  51 + if lastDeviceCollectionRecord == nil {
  52 + //log.Logger.Error("未找到上一条设备数据")
  53 + return nil, nil
  54 + }
  55 + }
  56 +
  57 + transactionContext, err := factory.CreateTransactionContext(nil)
  58 + if err != nil {
  59 + return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())
  60 + }
  61 + if err := transactionContext.StartTransaction(); err != nil {
  62 + return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())
  63 + }
  64 + defer func() {
  65 + transactionContext.RollbackTransaction()
  66 + }()
  67 +
41 var deviceCollectionRepository domain.DeviceCollectionRepository 68 var deviceCollectionRepository domain.DeviceCollectionRepository
42 if value, err := factory.CreateDeviceCollectionRepository(map[string]interface{}{ 69 if value, err := factory.CreateDeviceCollectionRepository(map[string]interface{}{
43 "transactionContext": transactionContext, 70 "transactionContext": transactionContext,
@@ -46,16 +73,49 @@ func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(c @@ -46,16 +73,49 @@ func (deviceCollectionService *DeviceCollectionService) CreateDeviceCollection(c
46 } else { 73 } else {
47 deviceCollectionRepository = value 74 deviceCollectionRepository = value
48 } 75 }
49 - if deviceCollection, err := deviceCollectionRepository.Save(newDeviceCollection); err != nil { 76 + deviceCollection, err := deviceCollectionRepository.Save(newDeviceCollection)
  77 + if err != nil {
50 return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error()) 78 return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
51 - } else {  
52 - if err := transactionContext.CommitTransaction(); err != nil {  
53 - return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 79 + }
  80 +
  81 + //处理设备数据
  82 + switch deviceCollection.DeviceType {
  83 + case domain.DeviceTypeBaoXianJi, domain.DeviceTypeChuanChuanJi, domain.DeviceTypeFengKouJi, domain.DeviceTypeFengXiangJi:
  84 + if v, ok := deviceCollection.Values["Count"]; ok {
  85 + curCount, errCurCount := strconv.Atoi(fmt.Sprintf("%v", v))
  86 + v, ok = lastDeviceCollectionRecord.Values["Count"]
  87 + if ok {
  88 + lastCount, errLastCount := strconv.Atoi(fmt.Sprintf("%v", v))
  89 + if errLastCount == nil && errCurCount == nil && lastCount <= curCount {
  90 + deviceCollection.Values["Count"] = curCount - lastCount
  91 + } else {
  92 + deviceCollection.Values["Count"] = 0
  93 + /*
  94 + 设备统计的数量超过一定范围会重置为0，特殊处理0操作
  95 + */
  96 + if lastCount > 10000000 && curCount < 1000 {
  97 + deviceCollection.Values["Count"] = curCount
  98 + }
  99 + }
  100 + } else {
  101 + deviceCollection.Values["Count"] = 0
  102 + }
54 } 103 }
55 - return map[string]interface{}{  
56 - "deviceCollection": deviceCollection,  
57 - }, nil 104 + break
58 } 105 }
  106 + err = domainService.SendWorkshopDeviceData(deviceCollection)
  107 + if err != nil {
  108 + log.Logger.Error("车间设备数据加入redis失败：" + err.Error())
  109 + return nil, application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  110 + }
  111 +
  112 + if err := transactionContext.CommitTransaction(); err != nil {
  113 + return nil, application.ThrowError(application.TRANSACTION_ERROR, err.Error())
  114 + }
  115 + return map[string]interface{}{
  116 + "deviceCollection": deviceCollection,
  117 + }, nil
  118 +
59 } 119 }
60 120
61 // 返回 121 // 返回
@@ -219,3 +279,47 @@ func NewDeviceCollectionService(options map[string]interface{}) *DeviceCollectio @@ -219,3 +279,47 @@ func NewDeviceCollectionService(options map[string]interface{}) *DeviceCollectio
219 newDeviceCollectionService := &DeviceCollectionService{} 279 newDeviceCollectionService := &DeviceCollectionService{}
220 return newDeviceCollectionService 280 return newDeviceCollectionService
221 } 281 }
  282 +
  283 +const DefaultReceiveSpan = 60 // 60 sec
  284 +
  285 +type DeviceDataInstance struct {
  286 + //deviceDataCache sync.Map
  287 + deviceDataLastTime sync.Map
  288 +}
  289 +
  290 +func NewDeviceDataInstance() *DeviceDataInstance {
  291 + return &DeviceDataInstance{
  292 + deviceDataLastTime: sync.Map{},
  293 + }
  294 +}
  295 +
  296 +func (d *DeviceDataInstance) Add(deviceSn, deviceType string, data interface{}) (*domain.DeviceCollection, error) {
  297 + // 获取数据上一次的
  298 + var v interface{}
  299 + var ok bool
  300 + var now = time.Now().Unix()
  301 + var t = now
  302 + if v, ok = d.deviceDataLastTime.Load(deviceSn); ok {
  303 + t = v.(int64)
  304 + } else {
  305 + d.deviceDataLastTime.Store(deviceSn, t)
  306 + redis.SaveDeviceRealTimeData(deviceSn, data, true)
  307 + return nil, errors.New(fmt.Sprintf("ingnore this record"))
  308 + }
  309 + // 60秒接受一次数据，暂时不用太多数据
  310 + if now < t+DefaultReceiveSpan {
  311 + return nil, errors.New(fmt.Sprintf("receive too fast wait %v sec", t+DefaultReceiveSpan-now))
  312 + }
  313 + // 从redis获取最后的数据进行处理
  314 + var result = &domain.DeviceCollection{}
  315 + if err := redis.GetDeviceRealTimeData(deviceSn, result); err != nil {
  316 + if err == domain.ErrorNotFound {
  317 + redis.SaveDeviceRealTimeData(deviceSn, data, true)
  318 + return nil, nil
  319 + }
  320 + return nil, err
  321 + }
  322 + //log.Logger.Debug("",map[string]interface{}{"t":t,"device":deviceSn})
  323 + d.deviceDataLastTime.Store(deviceSn, now)
  324 + return result, redis.SaveDeviceRealTimeData(deviceSn, data, false)
  325 +}
@@ -2,6 +2,7 @@ package query @@ -2,6 +2,7 @@ package query
2 2
3 import ( 3 import (
4 "fmt" 4 "fmt"
  5 + "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
5 "reflect" 6 "reflect"
6 "strings" 7 "strings"
7 8
@@ -31,18 +32,18 @@ type SearchProductGroupQuery struct { @@ -31,18 +32,18 @@ type SearchProductGroupQuery struct {
31 LineName string `cname:"生产线名称" json:"lineName,omitempty"` 32 LineName string `cname:"生产线名称" json:"lineName,omitempty"`
32 } 33 }
33 34
34 -func (listProductGroupQuery *SearchProductGroupQuery) Valid(validation *validation.Validation) {  
35 - //validation.SetError("CustomValid", "未实现的自定义认证") 35 +func (cmd *SearchProductGroupQuery) Valid(validation *validation.Validation) {
  36 + cmd.Offset, cmd.Limit = domain.Pagination(cmd.PageNumber, cmd.PageSize)
36 } 37 }
37 38
38 -func (listProductGroupQuery *SearchProductGroupQuery) ValidateQuery() error { 39 +func (cmd *SearchProductGroupQuery) ValidateQuery() error {
39 valid := validation.Validation{} 40 valid := validation.Validation{}
40 - b, err := valid.Valid(listProductGroupQuery) 41 + b, err := valid.Valid(cmd)
41 if err != nil { 42 if err != nil {
42 return err 43 return err
43 } 44 }
44 if !b { 45 if !b {
45 - elem := reflect.TypeOf(listProductGroupQuery).Elem() 46 + elem := reflect.TypeOf(cmd).Elem()
46 for _, validErr := range valid.Errors { 47 for _, validErr := range valid.Errors {
47 field, isExist := elem.FieldByName(validErr.Field) 48 field, isExist := elem.FieldByName(validErr.Field)
48 if isExist { 49 if isExist {
@@ -26,11 +26,12 @@ type DeviceYouZhaJi2 struct { @@ -26,11 +26,12 @@ type DeviceYouZhaJi2 struct {
26 26
27 // 串串机 27 // 串串机
28 type DeviceChuanChuanJi struct { 28 type DeviceChuanChuanJi struct {
29 - Count int64 `json:"Count"` // 生产计数：生产统计数量  
30 - Year string `json:"Year"` // 年  
31 - Month string `json:"Month"` // 月  
32 - Day string `json:"Day"` // 日  
33 - ProductType string `json:"ProductType"` // 产品类型：当前产品种类 29 + Count int64 `json:"Count"` // 生产计数：生产统计数量
  30 + Year int `json:"Year"` // 年
  31 + Month int `json:"Month"` // 月
  32 + Day int `json:"Day"` // 日
  33 + ProductType int `json:"ProductType"`
  34 + ProductType1 string `json:"ProductType1"` // 产品类型：当前产品种类
34 } 35 }
35 36
36 // 速冻线 37 // 速冻线
@@ -40,18 +41,20 @@ type DeviceSuDongXian struct { @@ -40,18 +41,20 @@ type DeviceSuDongXian struct {
40 41
41 // 封口机 42 // 封口机
42 type DeviceFengKouJi struct { 43 type DeviceFengKouJi struct {
43 - Count int64 `json:"Count"` // 生产计数：生产统计数量  
44 - Year string `json:"Year"` // 年  
45 - Month string `json:"Month"` // 月  
46 - Day string `json:"Day"` // 日  
47 - ProductType string `json:"ProductType"` // 产品类型：当前产品种类 44 + Count int64 `json:"Count"` // 生产计数：生产统计数量
  45 + Year int `json:"Year"` // 年
  46 + Month int `json:"Month"` // 月
  47 + Day int `json:"Day"` // 日
  48 + ProductType int `json:"ProductType"`
  49 + ProductType1 string `json:"ProductType1"` // 产品类型：当前产品种类
48 } 50 }
49 51
50 // 封箱机 52 // 封箱机
51 type DeviceFengXiangJi struct { 53 type DeviceFengXiangJi struct {
52 - Count int64 `json:"Count"` // 生产计数：生产统计数量  
53 - Year string `json:"Year"` // 年  
54 - Month string `json:"Month"` // 月  
55 - Day string `json:"Day"` // 日  
56 - ProductType string `json:"ProductType"` // 产品类型：当前产品种类 54 + Count int64 `json:"Count"` // 生产计数：生产统计数量
  55 + Year int `json:"Year"` // 年
  56 + Month int `json:"Month"` // 月
  57 + Day int `json:"Day"` // 日
  58 + ProductType int `json:"ProductType"`
  59 + ProductType1 string `json:"ProductType1"` // 产品类型：当前产品种类
57 } 60 }
@@ -10,7 +10,6 @@ import ( @@ -10,7 +10,6 @@ import (
10 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/repository" 10 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/repository"
11 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils" 11 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils"
12 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log" 12 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
13 - "strconv"  
14 "time" 13 "time"
15 ) 14 )
16 15
@@ -127,11 +126,11 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev @@ -127,11 +126,11 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev
127 if mBytes, err = json.Marshal(record.Values); err != nil { 126 if mBytes, err = json.Marshal(record.Values); err != nil {
128 return nil, err 127 return nil, err
129 } 128 }
130 - var formatDate = func(y, m, d string) (string, error) {  
131 - yd, _ := strconv.Atoi(y)  
132 - md, _ := strconv.Atoi(m)  
133 - dd, _ := strconv.Atoi(d)  
134 - t := time.Date(yd, time.Month(md), dd, 0, 0, 0, 0, time.Local) 129 + var formatDate = func(y, m, d int) (string, error) {
  130 + //yd, _ := strconv.Atoi(y)
  131 + //md, _ := strconv.Atoi(m)
  132 + //dd, _ := strconv.Atoi(d)
  133 + t := time.Date(y, time.Month(m), d, 0, 0, 0, 0, time.Local)
135 return t.Local().Format("2006-01-02"), nil 134 return t.Local().Format("2006-01-02"), nil
136 } 135 }
137 switch record.DeviceType { 136 switch record.DeviceType {
@@ -172,7 +171,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev @@ -172,7 +171,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev
172 break 171 break
173 } 172 }
174 data.Count = int(deviceChuanChuanJi.Count) 173 data.Count = int(deviceChuanChuanJi.Count)
175 - data.ProductType = deviceChuanChuanJi.ProductType 174 + data.ProductType = deviceChuanChuanJi.ProductType1
176 if data.Date, err = formatDate(deviceChuanChuanJi.Year, deviceChuanChuanJi.Month, deviceChuanChuanJi.Day); err != nil { 175 if data.Date, err = formatDate(deviceChuanChuanJi.Year, deviceChuanChuanJi.Month, deviceChuanChuanJi.Day); err != nil {
177 return nil, err 176 return nil, err
178 } 177 }
@@ -195,7 +194,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev @@ -195,7 +194,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev
195 break 194 break
196 } 195 }
197 data.Count = int(deviceFengKouJi.Count) 196 data.Count = int(deviceFengKouJi.Count)
198 - data.ProductType = deviceFengKouJi.ProductType 197 + data.ProductType = deviceFengKouJi.ProductType1
199 if data.Date, err = formatDate(deviceFengKouJi.Year, deviceFengKouJi.Month, deviceFengKouJi.Day); err != nil { 198 if data.Date, err = formatDate(deviceFengKouJi.Year, deviceFengKouJi.Month, deviceFengKouJi.Day); err != nil {
200 return nil, err 199 return nil, err
201 } 200 }
@@ -208,7 +207,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev @@ -208,7 +207,7 @@ func (ptr *PGWorkshopDataConsumeService) newDeviceRunningData(record *domain.Dev
208 break 207 break
209 } 208 }
210 data.Count = int(deviceFengXiangJi.Count) 209 data.Count = int(deviceFengXiangJi.Count)
211 - data.ProductType = deviceFengXiangJi.ProductType 210 + data.ProductType = deviceFengXiangJi.ProductType1
212 if data.Date, err = formatDate(deviceFengXiangJi.Year, deviceFengXiangJi.Month, deviceFengXiangJi.Day); err != nil { 211 if data.Date, err = formatDate(deviceFengXiangJi.Year, deviceFengXiangJi.Month, deviceFengXiangJi.Day); err != nil {
213 return nil, err 212 return nil, err
214 } 213 }
@@ -26,8 +26,8 @@ func (subscribeClient *SubscribeClient) options() *pahomqtt.ClientOptions { @@ -26,8 +26,8 @@ func (subscribeClient *SubscribeClient) options() *pahomqtt.ClientOptions {
26 opts.SetPassword(constant.MQTT_PASSWORD) 26 opts.SetPassword(constant.MQTT_PASSWORD)
27 opts.SetKeepAlive(2 * time.Second) 27 opts.SetKeepAlive(2 * time.Second)
28 opts.SetPingTimeout(1 * time.Second) 28 opts.SetPingTimeout(1 * time.Second)
29 - opts.CleanSession = false  
30 - //opts.SetClientID("test") 29 + //opts.CleanSession = false
  30 + opts.SetClientID(constant.SERVICE_NAME)
31 //opts.Order = true 31 //opts.Order = true
32 return opts 32 return opts
33 } 33 }
  1 +package redis
  2 +
  3 +import (
  4 + "fmt"
  5 + "github.com/go-redis/redis"
  6 + "github.com/linmadan/egglib-go/utils/json"
  7 + "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
  8 + "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
  9 +)
  10 +
  11 +/*
  12 + 保存实时数据
  13 +
  14 + device 设备号
  15 + record 记录
  16 + saveNotExists true:不存在才保存 false:直接覆盖
  17 +*/
  18 +func SaveDeviceRealTimeData(device string, record interface{}, saveNotExists bool) error {
  19 + client := GetRedis()
  20 + recordData, err := json.Marshal(record)
  21 + if err != nil {
  22 + return err
  23 + }
  24 + key := DeviceRealTimeDataKey(device)
  25 + // 已存在的不做保存
  26 + if saveNotExists {
  27 + if exists, err := client.Exists(key).Result(); exists > 0 && err == nil {
  28 + return nil
  29 + }
  30 + }
  31 + result := client.Set(key, recordData, 0)
  32 + _, err = result.Result()
  33 + return err
  34 +}
  35 +
  36 +/*
  37 + 获取实时数据
  38 +*/
  39 +func GetDeviceRealTimeData(device string, val interface{}) error {
  40 + client := GetRedis()
  41 + key := DeviceRealTimeDataKey(device)
  42 + result := client.Get(key)
  43 + data, err := result.Bytes()
  44 + if err == redis.Nil {
  45 + return domain.ErrorNotFound
  46 + }
  47 + if err = json.Unmarshal(data, val); err != nil {
  48 + return err
  49 + }
  50 + return nil
  51 +}
  52 +
  53 +func DeviceRealTimeDataKey(deviceType string) string {
  54 + str := fmt.Sprintf("%v:device-realtime-data:%v", constant.CACHE_PREFIX, deviceType)
  55 + return str
  56 +}
@@ -2,7 +2,6 @@ package mqtt @@ -2,7 +2,6 @@ package mqtt
2 2
3 import ( 3 import (
4 "encoding/json" 4 "encoding/json"
5 - "fmt"  
6 pahomqtt "github.com/eclipse/paho.mqtt.golang" 5 pahomqtt "github.com/eclipse/paho.mqtt.golang"
7 "github.com/linmadan/egglib-go/utils/tool_funs" 6 "github.com/linmadan/egglib-go/utils/tool_funs"
8 "github.com/tidwall/gjson" 7 "github.com/tidwall/gjson"
@@ -11,157 +10,137 @@ import ( @@ -11,157 +10,137 @@ import (
11 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant" 10 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/constant"
12 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain" 11 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/domain"
13 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/mqtt" 12 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/mqtt"
14 - "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/redis"  
15 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils" 13 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/infrastructure/utils"
16 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log" 14 "gitlab.fjmaimaimai.com/allied-creation/allied-creation-manufacture/pkg/log"
17 "time" 15 "time"
18 ) 16 )
19 17
20 func Start() { 18 func Start() {
21 - mqtt.StartSubscribe(constant.MQTT_TOPIC, func(client pahomqtt.Client, message pahomqtt.Message) {  
22 - payload := make(map[string]interface{})  
23 - err := json.Unmarshal(message.Payload(), &payload)  
24 - if err != nil {  
25 - log.Logger.Error("车间数据json解析失败：" + err.Error())  
26 - return  
27 - }  
28 - if workShop, ok := payload["WorkShop"]; ok {  
29 - for key, item := range payload {  
30 - if key == "WorkShop" {  
31 - continue  
32 - }  
33 - mBytes, err := json.Marshal(item)  
34 - if err != nil {  
35 - continue  
36 - }  
37 - deviceCollection := &domain.DeviceCollection{  
38 - WorkShopName: workShop.(string),  
39 - DeviceSn: key,  
40 - CollectionTime: time.Now(),  
41 - StartupStatus: gjson.Get(string(message.Payload()), key+".StartupState").Int(),  
42 - ComStatus: gjson.Get(string(message.Payload()), key+".ComStatus").Int(),  
43 - }  
44 - if utils.SubStr(key, 0, 4) == domain.DeviceTypeMianBaoXieJi {  
45 - deviceCollection.DeviceType = domain.DeviceTypeMianBaoXieJi  
46 - } else {  
47 - deviceType := utils.SubStr(key, 0, 3)  
48 - deviceCollection.DeviceType = deviceType  
49 - switch deviceType {  
50 - //包馅机  
51 - case domain.DeviceTypeBaoXianJi:  
52 - deviceBaoXianJi := &domain.DeviceBaoXianJi{}  
53 - err = json.Unmarshal(mBytes, deviceBaoXianJi)  
54 - if err != nil {  
55 - continue  
56 - }  
57 - deviceCollection.Values = tool_funs.SimpleStructToMap(deviceBaoXianJi)  
58 - break  
59 - //油炸机  
60 - case domain.DeviceTypeYouZhaJi1:  
61 - deviceYouZhaJi := &domain.DeviceYouZhaJi{}  
62 - err = json.Unmarshal(mBytes, deviceYouZhaJi)  
63 - if err != nil {  
64 - continue  
65 - }  
66 - deviceCollection.Values = tool_funs.SimpleStructToMap(deviceYouZhaJi)  
67 - break  
68 - //油炸机  
69 - case domain.DeviceTypeYouZhaJi2:  
70 - deviceYouZhaJi2 := &domain.DeviceYouZhaJi2{}  
71 - err = json.Unmarshal(mBytes, deviceYouZhaJi2)  
72 - if err != nil {  
73 - continue  
74 - }  
75 - deviceCollection.Values = tool_funs.SimpleStructToMap(deviceYouZhaJi2)  
76 - break  
77 - //串串机  
78 - case domain.DeviceTypeChuanChuanJi:  
79 - deviceChuanChuanJi := &domain.DeviceChuanChuanJi{}  
80 - err = json.Unmarshal(mBytes, deviceChuanChuanJi)  
81 - if err != nil {  
82 - continue  
83 - }  
84 - deviceCollection.Values = tool_funs.SimpleStructToMap(deviceChuanChuanJi)  
85 - break  
86 - //速冻线  
87 - case domain.DeviceTypeSuDongXian:  
88 - deviceSuDongXian := &domain.DeviceSuDongXian{}  
89 - err = json.Unmarshal(mBytes, deviceSuDongXian)  
90 - if err != nil {  
91 - continue  
92 - }  
93 - deviceCollection.Values = tool_funs.SimpleStructToMap(deviceSuDongXian)  
94 - break  
95 - //封口机  
96 - case domain.DeviceTypeFengKouJi:  
97 - deviceFengKouJi := &domain.DeviceFengKouJi{}  
98 - err = json.Unmarshal(mBytes, deviceFengKouJi)  
99 - if err != nil {  
100 - continue  
101 - }  
102 - deviceCollection.Values = tool_funs.SimpleStructToMap(deviceFengKouJi)  
103 - break  
104 - //封箱机  
105 - case domain.DeviceTypeFengXiangJi:  
106 - deviceFengXiangJi := &domain.DeviceFengXiangJi{}  
107 - err = json.Unmarshal(mBytes, deviceFengXiangJi)  
108 - if err != nil {  
109 - continue  
110 - }  
111 - deviceCollection.Values = tool_funs.SimpleStructToMap(deviceFengXiangJi)  
112 - break  
113 - //打浆机 //面包屑机  
114 - case domain.DeviceTypeDaJiangJi:  
115 - default: 19 + mqtt.StartSubscribe(constant.MQTT_TOPIC, OnReceiveData)
  20 +}
  21 +
  22 +func OnReceiveData(client pahomqtt.Client, message pahomqtt.Message) {
  23 + payload := make(map[string]interface{})
  24 + err := json.Unmarshal(message.Payload(), &payload)
  25 + if err != nil {
  26 + log.Logger.Error("车间数据json解析失败：" + err.Error())
  27 + return
  28 + }
  29 + if workShop, ok := payload["WorkShop"]; ok {
  30 + for key, item := range payload {
  31 + if key == "WorkShop" {
  32 + continue
  33 + }
  34 + mBytes, err := json.Marshal(item)
  35 + if err != nil {
  36 + continue
  37 + }
  38 + deviceCollection := &domain.DeviceCollection{
  39 + WorkShopName: workShop.(string),
  40 + DeviceSn: key,
  41 + CollectionTime: time.Now(),
  42 + StartupStatus: gjson.Get(string(message.Payload()), key+".StartupState").Int(),
  43 + ComStatus: gjson.Get(string(message.Payload()), key+".ComStatus").Int(),
  44 + }
  45 + if utils.SubStr(key, 0, 4) == domain.DeviceTypeMianBaoXieJi {
  46 + deviceCollection.DeviceType = domain.DeviceTypeMianBaoXieJi
  47 + } else {
  48 + deviceType := utils.SubStr(key, 0, 3)
  49 + deviceCollection.DeviceType = deviceType
  50 + switch deviceType {
  51 + //包馅机
  52 + case domain.DeviceTypeBaoXianJi:
  53 + deviceBaoXianJi := &domain.DeviceBaoXianJi{}
  54 + err = json.Unmarshal(mBytes, deviceBaoXianJi)
  55 + if err != nil {
  56 + continue
116 } 57 }
117 - deviceCollectionService := service.NewDeviceCollectionService(nil)  
118 - resp, err := deviceCollectionService.CreateDeviceCollection(&command.CreateDeviceCollectionCommand{  
119 - WorkShopName: deviceCollection.WorkShopName,  
120 - StartupStatus: deviceCollection.StartupStatus,  
121 - CollectionTime: deviceCollection.CollectionTime,  
122 - DeviceSn: deviceCollection.DeviceSn,  
123 - DeviceType: deviceCollection.DeviceType,  
124 - ComStatus: deviceCollection.ComStatus,  
125 - Values: deviceCollection.Values,  
126 - }) 58 + deviceCollection.Values = tool_funs.SimpleStructToMap(deviceBaoXianJi)
  59 + break
  60 + //油炸机
  61 + case domain.DeviceTypeYouZhaJi1:
  62 + deviceYouZhaJi := &domain.DeviceYouZhaJi{}
  63 + err = json.Unmarshal(mBytes, deviceYouZhaJi)
127 if err != nil { 64 if err != nil {
128 continue 65 continue
129 } 66 }
130 -  
131 - result := resp.(map[string]interface{})  
132 - if deviceCollectionResult, ok := result["deviceCollection"]; ok {  
133 - fmt.Println(deviceCollectionResult)  
134 - deviceCollection.DeviceCollectionId = deviceCollectionResult.(*domain.DeviceCollection).DeviceCollectionId  
135 - workShopBytes, err := json.Marshal(deviceCollection)  
136 - if err != nil {  
137 - continue  
138 - }  
139 - err = redis.GetRedis().LPush(constant.REDIS_WORKSHOP_KEY, string(workShopBytes)).Err()  
140 - if err != nil {  
141 - log.Logger.Error("车间设备数据加入redis失败：" + err.Error())  
142 - }  
143 -  
144 - //workShopBytes, err := json.Marshal(deviceCollection)  
145 - //if err != nil {  
146 - // continue  
147 - //}  
148 - //err = redis.GetRedis().LPush(constant.REDIS_WORKSHOP_KEY, string(workShopBytes)).Err()  
149 - //if err != nil {  
150 - // log.Logger.Error("车间设备数据加入redis失败：" + err.Error())  
151 - //}  
152 - //err = domainService.SendWorkshopDeviceData(deviceCollection)  
153 - //if err != nil {  
154 - // log.Logger.Error("车间设备数据加入redis失败：" + err.Error())  
155 - //  
156 - //} 67 + deviceCollection.Values = tool_funs.SimpleStructToMap(deviceYouZhaJi)
  68 + break
  69 + //油炸机
  70 + case domain.DeviceTypeYouZhaJi2:
  71 + deviceYouZhaJi2 := &domain.DeviceYouZhaJi2{}
  72 + err = json.Unmarshal(mBytes, deviceYouZhaJi2)
  73 + if err != nil {
  74 + continue
157 } 75 }
  76 + deviceCollection.Values = tool_funs.SimpleStructToMap(deviceYouZhaJi2)
  77 + break
  78 + //串串机
  79 + case domain.DeviceTypeChuanChuanJi:
  80 + deviceChuanChuanJi := &domain.DeviceChuanChuanJi{}
  81 + err = json.Unmarshal(mBytes, deviceChuanChuanJi)
  82 + if err != nil {
  83 + log.Logger.Error(err.Error())
  84 + continue
  85 + }
  86 + deviceCollection.Values = tool_funs.SimpleStructToMap(deviceChuanChuanJi)
  87 + break
  88 + //速冻线
  89 + case domain.DeviceTypeSuDongXian:
  90 + deviceSuDongXian := &domain.DeviceSuDongXian{}
  91 + err = json.Unmarshal(mBytes, deviceSuDongXian)
  92 + if err != nil {
  93 + continue
  94 + }
  95 + deviceCollection.Values = tool_funs.SimpleStructToMap(deviceSuDongXian)
  96 + break
  97 + //封口机
  98 + case domain.DeviceTypeFengKouJi:
  99 + deviceFengKouJi := &domain.DeviceFengKouJi{}
  100 + err = json.Unmarshal(mBytes, deviceFengKouJi)
  101 + if err != nil {
  102 + continue
  103 + }
  104 + deviceCollection.Values = tool_funs.SimpleStructToMap(deviceFengKouJi)
  105 + break
  106 + //封箱机
  107 + case domain.DeviceTypeFengXiangJi:
  108 + deviceFengXiangJi := &domain.DeviceFengXiangJi{}
  109 + err = json.Unmarshal(mBytes, deviceFengXiangJi)
  110 + if err != nil {
  111 + continue
  112 + }
  113 + deviceCollection.Values = tool_funs.SimpleStructToMap(deviceFengXiangJi)
  114 + break
  115 + //打浆机 //面包屑机
  116 + case domain.DeviceTypeDaJiangJi:
  117 + case domain.DeviceTypeChuanChuanJi:
  118 + default:
  119 + }
  120 + if deviceType != domain.DeviceTypeChuanChuanJi {
  121 + continue
158 } 122 }
159 - log.Logger.Info("MQTT", map[string]interface{}{  
160 - "Topic": message.Topic(),  
161 - "MessageId": message.MessageID(),  
162 - "Message": payload, 123 + // 发送数据
  124 + deviceCollectionService := service.NewDeviceCollectionService(nil)
  125 + _, err := deviceCollectionService.CreateDeviceCollection(&command.CreateDeviceCollectionCommand{
  126 + WorkShopName: deviceCollection.WorkShopName,
  127 + StartupStatus: deviceCollection.StartupStatus,
  128 + CollectionTime: deviceCollection.CollectionTime,
  129 + DeviceSn: deviceCollection.DeviceSn,
  130 + DeviceType: deviceCollection.DeviceType,
  131 + ComStatus: deviceCollection.ComStatus,
  132 + Values: deviceCollection.Values,
  133 + PreCheck: true,
163 }) 134 })
  135 + if err != nil {
  136 + continue
  137 + }
164 } 138 }
165 } 139 }
166 - }) 140 + log.Logger.Info("MQTT", map[string]interface{}{
  141 + "Topic": message.Topic(),
  142 + "MessageId": message.MessageID(),
  143 + "Message": payload,
  144 + })
  145 + }
167 } 146 }