停止消费时关闭tcp连接

zhangqian

2023-10-20 7276ab65576ec73b439a40d7f1a3035a534b968c

 nsq/msg_handler.go

@@ -79,6 +79,7 @@
      }

      err = model.WithTransaction(func(db *gorm.DB) error {
         var err error
         if oldWorkOrder != nil && oldWorkOrder.WorkOrderID != "" {
            if err = model.NewOrderSearch(db).SetId(oldWorkOrder.ID).Updates(&orderRecord); err != nil {
               return err
@@ -93,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
      })
@@ -213,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).
@@ -225,7 +235,6 @@
      if err != nil {
         return err
      }
      processModel.IsNew = true
      return model.NewProcessModelSearch().SetOrm(db).Create(&processModel)
   })

@@ -264,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 {
@@ -291,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)
@@ -338,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
}