| | |
| | | |
| | | return nil |
| | | } |
| | | |
| | | type PullDataResponse struct { |
| | | Topic string |
| | | } |
| | | |
| | | func (slf *PullDataResponse) HandleMessage(data []byte) (err error) { |
| | | logx.Infof("get a pull data response message :%s", data) |
| | | var pullDataResponse common.MsgPullDataResponse |
| | | err = json.Unmarshal(data, &pullDataResponse) |
| | | if err != nil { |
| | | logx.Infof("unmarshal msg err :%s", err) |
| | | return err |
| | | } |
| | | switch pullDataResponse.DataType { |
| | | case common.PullDataTypeProcessModel: |
| | | err = slf.DealProcessModelData(pullDataResponse.Data) |
| | | |
| | | } |
| | | if err != nil { |
| | | logx.Infof("process pull data err :%s", err) |
| | | return err |
| | | } |
| | | return nil |
| | | } |
| | | |
| | | func (slf *PullDataResponse) DealProcessModelData(data interface{}) error { |
| | | var processModels []*model.ProcessModel |
| | | err := mapstructure.Decode(data, &processModels) |
| | | if err != nil { |
| | | return err |
| | | } |
| | | numbers := make([]string, 0, len(processModels)) |
| | | for _, processModel := range processModels { |
| | | numbers = append(numbers, processModel.Number) |
| | | } |
| | | existsProcessModels, err := model.NewProcessModelSearch().SetIsNew(true).SetNumbers(numbers).FindNotTotal() |
| | | if err != nil { |
| | | return err |
| | | } |
| | | |
| | | existsProcessModelsMap := make(map[string]struct{}, len(existsProcessModels)) |
| | | for _, processModel := range existsProcessModels { |
| | | existsProcessModelsMap[processModel.Number] = struct{}{} |
| | | } |
| | | |
| | | for _, processModel := range processModels { |
| | | if _, exists := existsProcessModelsMap[processModel.Number]; exists { |
| | | continue |
| | | } |
| | | err = model.WithTransaction(func(db *gorm.DB) error { |
| | | err = model.NewProcessModelSearch().SetOrm(db).SetProcedure(processModel.Procedure).SetProduct(processModel.Product).SetIsNew(true).UpdateByMap(map[string]interface{}{"is_new": 0}) |
| | | if err != nil { |
| | | return err |
| | | } |
| | | processModel.IsNew = true |
| | | return model.NewProcessModelSearch().SetOrm(db).Create(processModel) |
| | | }) |
| | | if err != nil { |
| | | return err |
| | | } |
| | | } |
| | | return nil |
| | | } |