ERP_storage/Nats/Nats.go

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()
			}
		}

	})

}