From 7276ab65576ec73b439a40d7f1a3035a534b968c Mon Sep 17 00:00:00 2001
From: zhangqian <zhangqian@123.com>
Date: 星期五, 20 十月 2023 19:12:40 +0800
Subject: [PATCH] 停止消费时关闭tcp连接
---
nsq/msg_handler.go | 55 ++++++++++++++++++++++++++++++++++++++++++++++++-------
1 files changed, 48 insertions(+), 7 deletions(-)
diff --git a/nsq/msg_handler.go b/nsq/msg_handler.go
index f153270..3e1f160 100644
--- a/nsq/msg_handler.go
+++ b/nsq/msg_handler.go
@@ -94,7 +94,9 @@
if err = model.NewOrderSearch(db).Create(&orderRecord); err != nil {
return err
}
- return model.NewProceduresSearch(db).CreateBatch(procedureRecords)
+ if err = model.NewProceduresSearch(db).CreateBatch(procedureRecords); err != nil {
+ return err
+ }
}
return nil
})
@@ -214,11 +216,18 @@
func (slf *ProcessParamsSync) HandleMessage(data []byte) (err error) {
logx.Infof("get an process params sync message :%s", data)
- var processModel model.ProcessModel
- err = json.Unmarshal(data, &processModel)
+ var processModelMsg common.ResponseProcessParams
+ err = json.Unmarshal(data, &processModelMsg)
if err != nil {
logx.Infof("unmarshal process params sync err :%s", err)
return err
+ }
+ processModel := model.ProcessModel{
+ Number: processModelMsg.Number,
+ Product: processModelMsg.Product,
+ Procedure: processModelMsg.Procedure,
+ Params: processModelMsg.Params,
+ IsNew: true,
}
err = model.WithTransaction(func(db *gorm.DB) error {
err = model.NewProcessModelSearch().SetOrm(db).SetProduct(processModel.Product).SetProcedure(processModel.Procedure).
@@ -226,7 +235,6 @@
if err != nil {
return err
}
- processModel.IsNew = true
return model.NewProcessModelSearch().SetOrm(db).Create(&processModel)
})
@@ -265,8 +273,7 @@
if err == gorm.ErrRecordNotFound {
err = model.NewDeviceSearch().Create(deviceRecord)
} else {
- deviceRecord.ID = oldRecord.ID
- err = model.NewDeviceSearch().Save(deviceRecord)
+ err = model.NewDeviceSearch().SetID(oldRecord.ID).Save(deviceRecord)
}
if err != nil {
@@ -292,7 +299,8 @@
switch pullDataResponse.DataType {
case common.PullDataTypeProcessModel:
err = slf.DealProcessModelData(pullDataResponse.Data)
-
+ case common.PullDataTypeDevice:
+ err = slf.DealDeviceData(pullDataResponse.Data)
}
if err != nil {
logx.Infof("process pull data err :%s", err)
@@ -339,3 +347,36 @@
}
return nil
}
+
+func (slf *PullDataResponse) DealDeviceData(data interface{}) error {
+ var devices []*model.Device
+ err := mapstructure.Decode(data, &devices)
+ if err != nil {
+ return err
+ }
+ numbers := make([]string, 0, len(devices))
+ for _, item := range devices {
+ numbers = append(numbers, item.DeviceID)
+ }
+ existsDevices, err := model.NewDeviceSearch().SetDeviceIds(numbers).FindNotTotal()
+ if err != nil {
+ return err
+ }
+
+ existsDeviceMap := make(map[string]*model.Device, len(existsDevices))
+ for _, device := range existsDevices {
+ existsDeviceMap[device.DeviceID] = device
+ }
+
+ for _, device := range devices {
+ if oldDevice, exists := existsDeviceMap[device.DeviceID]; exists {
+ if oldDevice.ExtChannelAmount != device.ExtChannelAmount || //todo to be continued
+ oldDevice.Procedures != device.Procedures {
+ err = model.NewDeviceSearch().SetDeviceId(device.DeviceID).Save(device)
+ }
+ } else {
+ err = model.NewDeviceSearch().Create(device)
+ }
+ }
+ return err
+}
--
Gitblit v1.8.0