ERP_storage/Nats/Nats.go

package Nats

import (
	"ERP_storage/conf"
	"ERP_storage/models/Account"
	"ERP_storage/models/Contract"
	"encoding/json"
	"fmt"

	"github.com/astaxie/beego/logs"
	"github.com/beego/beego/v2/adapter/orm"
	"github.com/nats-io/nats.go"
	"github.com/vmihailenco/msgpack/v5"
	menulibs "gogs.baozhida.cn/zoie/ERP_libs/Menu"
	powerlibs "gogs.baozhida.cn/zoie/ERP_libs/Power"
	"gogs.baozhida.cn/zoie/ERP_libs/lib"
)

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

	})

	// ========== 菜单相关NATS服务 ==========
	// 请求-响应 添加菜单
	_, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Add_Menu", func(m *nats.Msg) {
		type T_S struct {
			Menu menulibs.Menu
			APIs string // API JSON字符串
		}
		type T_R struct {
			Code int16  `xml:"Code"`
			Msg  string `xml:"Msg"`
			Data int    `xml:"Data"`
		}

		var t_S T_S
		var t_R T_R

		err := msgpack.Unmarshal(m.Data, &t_S)
		if err != nil {
			t_R.Code = 202
			t_R.Msg = "msgpack unmarshal err!"
			b, _ := msgpack.Marshal(&t_R)
			_ = Nats.Publish(m.Reply, b)
			return
		}

		o := orm.NewOrm()
		MenuDao := menulibs.NewMenu(o)
		id64, err := MenuDao.Add_Menu(t_S.Menu)
		if err != nil {
			t_R.Code = 202
			t_R.Msg = "添加菜单失败!"
			b, _ := msgpack.Marshal(&t_R)
			_ = Nats.Publish(m.Reply, b)
			return
		}
		id := int(id64)

		// 处理API JSON字符串
		if len(t_S.APIs) > 0 {
			var apiList []struct {
				T_name   string `json:"T_name"`
				T_uri    string `json:"T_uri"`
				T_method string `json:"T_method"`
			}
			err = json.Unmarshal([]byte(t_S.APIs), &apiList)
			if err == nil && len(apiList) > 0 {
				var apis []menulibs.API
				for _, api := range apiList {
					if len(api.T_method) == 0 {
						api.T_method = "POST"
					}
					apis = append(apis, menulibs.API{
						T_Menu_Id: id,
						T_name:    api.T_name,
						T_uri:     api.T_uri,
						T_method:  api.T_method,
						T_enable:  1,
					})
				}
				APIDao := menulibs.NewAPI(o, Account.RedisCache_API)
				_, _ = APIDao.InsertMulti_API(apis)
			}
		}

		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_Menu", func(m *nats.Msg) {
		type T_S struct {
			Menu menulibs.Menu
			Cols []string // 需要更新的字段
			APIs string   // API JSON字符串
		}
		type T_R struct {
			Code int16  `xml:"Code"`
			Msg  string `xml:"Msg"`
			Data int    `xml:"Data"`
		}

		var t_S T_S
		var t_R T_R

		err := msgpack.Unmarshal(m.Data, &t_S)
		if err != nil {
			t_R.Code = 202
			t_R.Msg = "msgpack unmarshal err!"
			b, _ := msgpack.Marshal(&t_R)
			_ = Nats.Publish(m.Reply, b)
			return
		}

		o := orm.NewOrm()
		MenuDao := menulibs.NewMenu(o)
		APIDao := menulibs.NewAPI(o, Account.RedisCache_API)

		// 更新菜单
		if len(t_S.Cols) > 0 {
			_, err = MenuDao.Update_Menu(t_S.Menu, t_S.Cols...)
			if err != nil {
				t_R.Code = 202
				t_R.Msg = "更新菜单失败!"
				b, _ := msgpack.Marshal(&t_R)
				_ = Nats.Publish(m.Reply, b)
				return
			}
		}

		// 处理API JSON字符串
		if len(t_S.APIs) > 0 {
			// 先删除旧的API（软删除）
			_, _ = APIDao.Delete_API_ByMenuId(t_S.Menu.Id)

			var apiList []struct {
				T_name   string `json:"T_name"`
				T_uri    string `json:"T_uri"`
				T_method string `json:"T_method"`
			}
			err = json.Unmarshal([]byte(t_S.APIs), &apiList)
			if err == nil && len(apiList) > 0 {
				var apis []menulibs.API
				for _, api := range apiList {
					if len(api.T_method) == 0 {
						api.T_method = "POST"
					}
					apis = append(apis, menulibs.API{
						T_Menu_Id: t_S.Menu.Id,
						T_name:    api.T_name,
						T_uri:     api.T_uri,
						T_method:  api.T_method,
						T_enable:  1,
					})
				}
				_, _ = APIDao.InsertMulti_API(apis)
			}
		}

		t_R.Code = 200
		t_R.Msg = "ok"
		t_R.Data = t_S.Menu.Id
		b, _ := msgpack.Marshal(&t_R)
		_ = Nats.Publish(m.Reply, b)
	})

	// 请求-响应 删除菜单
	_, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Delete_Menu", func(m *nats.Msg) {
		type T_R struct {
			Code int16  `xml:"Code"`
			Msg  string `xml:"Msg"`
			Data int    `xml:"Data"`
		}

		var menu menulibs.Menu
		var t_R T_R

		err := msgpack.Unmarshal(m.Data, &menu)
		if err != nil {
			t_R.Code = 202
			t_R.Msg = "msgpack unmarshal err!"
			b, _ := msgpack.Marshal(&t_R)
			_ = Nats.Publish(m.Reply, b)
			return
		}

		o := orm.NewOrm()
		MenuDao := menulibs.NewMenu(o)
		_, err = MenuDao.Delete_Menu(menu)
		if err != nil {
			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.Id
		b, _ := msgpack.Marshal(&t_R)
		_ = Nats.Publish(m.Reply, b)
	})

	// 请求-响应 根据ID获取菜单
	_, _ = Nats.Subscribe(conf.NatsSubj_Prefix+conf.Sys_Name+"_Read_Menu_ById", func(m *nats.Msg) {
		type T_R struct {
			Code int16         `xml:"Code"`
			Msg  string        `xml:"Msg"`
			Data menulibs.Menu `xml:"Data"`
		}

		var id int
		var t_R T_R

		err := msgpack.Unmarshal(m.Data, &id)
		if err != nil {
			t_R.Code = 202
			t_R.Msg = "msgpack unmarshal err!"
			b, _ := msgpack.Marshal(&t_R)
			_ = Nats.Publish(m.Reply, b)
			return
		}

		o := orm.NewOrm()
		MenuDao := menulibs.NewMenu(o)
		menu, err := MenuDao.Read_Menu_ById(id)
		if err != nil {
			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)
	})

}