package Nats import ( "ERP_storage/conf" "ERP_storage/models/Account" "ERP_storage/models/Contract" "fmt" menulibs "git.baozhida.cn/ERP_libs/Menu" powerlibs "git.baozhida.cn/ERP_libs/Power" "git.baozhida.cn/ERP_libs/lib" "github.com/astaxie/beego/logs" "github.com/beego/beego/v2/adapter/orm" "github.com/nats-io/nats.go" "github.com/vmihailenco/msgpack/v5" ) var Nats *nats.Conn func init() { fmt.Println("============Nats init============") var err error // 连接Nats服务器 Nats, err = nats.Connect("nats://" + conf.NatsServer_Url) if err != nil { fmt.Println("nats 连接失败!") panic(err) } fmt.Println("nats OK!") go NatsInit() } func NatsInit() { // 发布-订阅 模式,异步订阅 test1 _, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Read_Menu_List", func(m *nats.Msg) { var t_R lib.JSONS o := orm.NewOrm() MenuDao := menulibs.NewMenu(o) menu, err := MenuDao.Read_Menu_List() if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = "获取失败!" b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } t_R.Code = 200 t_R.Msg = "ok" t_R.Data = menu b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) }) _, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Read_User_Bind_Menu_List", func(m *nats.Msg) { var t_R lib.JSONS o := orm.NewOrm() powerDao := powerlibs.NewPower(o) power, err := powerDao.Read_Power_ByT_id(string(m.Data)) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } MenuDao := menulibs.NewMenu(o) menu, err := MenuDao.Read_Menu_List_ByPower_T_Menu(power.T_menu) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } t_R.Code = 200 t_R.Msg = "ok" t_R.Data = menu b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) }) _, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Add_Power", func(m *nats.Msg) { var t_Req powerlibs.Power var t_R lib.JSONS err := msgpack.Unmarshal(m.Data, &t_Req) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } fmt.Printf("ERP_storage_Add_Power message: %+v\n", t_Req) o := orm.NewOrm() powerDao := powerlibs.NewPower(o) id, err := powerDao.Add_Power(t_Req) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } t_R.Code = 200 t_R.Msg = "ok" t_R.Data = id b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) }) _, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Read_Power_ByT_id", func(m *nats.Msg) { fmt.Printf("ERP_storage_Read_Power_ByT_id message: %+v\n", string(m.Data)) var t_R lib.JSONS o := orm.NewOrm() powerDao := powerlibs.NewPower(o) power, err := powerDao.Read_Power_ByT_id(string(m.Data)) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } t_R.Code = 200 t_R.Msg = "ok" t_R.Data = power b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) }) _, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Update_Power", func(m *nats.Msg) { var t_Req powerlibs.Power var t_R lib.JSONS err := msgpack.Unmarshal(m.Data, &t_Req) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } fmt.Printf("ERP_storage_Update_Power message: %+v\n", t_Req) o := orm.NewOrm() powerDao := powerlibs.NewPower(o) id, err := powerDao.Update_Power(t_Req) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } APIDao := menulibs.NewAPI(o, Account.RedisCache_API) APIDao.Redis_API_DelK(t_Req.T_id) t_R.Code = 200 t_R.Msg = "ok" t_R.Data = id b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) }) _, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Delete_Power", func(m *nats.Msg) { var t_Req powerlibs.Power var t_R lib.JSONS err := msgpack.Unmarshal(m.Data, &t_Req) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } fmt.Printf("ERP_storage_Delete_Power message: %+v\n", t_Req) o := orm.NewOrm() powerDao := powerlibs.NewPower(o) id, err := powerDao.Delete_Power(t_Req) if err != nil { logs.Error("Mats", lib.FuncName(), err) t_R.Code = 202 t_R.Msg = err.Error() b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) return } APIDao := menulibs.NewAPI(o, Account.RedisCache_API) APIDao.Redis_API_DelK(t_Req.T_id) t_R.Code = 200 t_R.Msg = "ok" t_R.Data = id b, _ := msgpack.Marshal(&t_R) _ = Nats.Publish(m.Reply, b) }) // 更新合同状态 _, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Update_Contract_T_out", func(m *nats.Msg) { fmt.Printf(conf.Sys_Name+"_Update_Contract_T_out message: %v\n", string(m.Data)) T_number := string(m.Data) o := orm.NewOrm() o.Begin() ContractDao := Contract.NewContract(o) ContractProductDao := Contract.NewContractProduct(o) // 1、添加出库单 contract, _ := ContractDao.Read_Contract_ByT_number(T_number) T_out := 2 // 查询合同产品清单是否全部都为已出库 state := ContractProductDao.Read_ContractProduct_T_State_List(T_number) if state == 1 { T_out = 1 } if state == 3 { T_out = 3 } if T_out != contract.T_out { err := ContractDao.Update_Contract(contract, "T_out") if err != nil { o.Rollback() } } }) }