MTP20_ACCESS/service/mq/mq.go

package mq

import (
	"errors"
	"fmt"
	"mtp20access/global"
	commonRequest "mtp20access/model/common/request"
	rsp "mtp20access/model/common/response"
	"mtp20access/model/mq/request"
	"mtp20access/model/mq/response"
	"time"

	"github.com/gin-gonic/gin"
	"go.uber.org/zap"
)

// SendMQ 向总线发送信息
func SendMQ(c *gin.Context, req *request.MQBodyReq) (err error) {
	// 获取请求账号信息
	s, exists := c.Get("claims")
	if !exists {
		err = errors.New("获取请求账号信息异常")
		global.M2A_LOG.Error(err.Error(), zap.Error(err))
		return
	}
	claims := s.(*commonRequest.CustomClaims)

	// 获取登录账户信息
	client, exists := global.M2A_Clients[claims.SessionID]
	if !exists {
		err = errors.New("获取登录账户信息异常")
		global.M2A_LOG.Error(err.Error(), zap.Error(err))
		return
	}

	// 将客户端请求的数据转换成总线使用的Protobuf
	serialNumber := client.GetSerialNumber()
	bytes, err := req.GetProtoBytes(&serialNumber)
	if err != nil {
		return
	}
	if bytes == nil {
		err = errors.New("请求信息序列化失败")
		return
	}

	// 创建异步任务
	key := fmt.Sprintf("%v_%v_%v", client.SessionID, req.FunCodeRsp, serialNumber)
	asyncTask := global.AsyncTask{
		FuncodeRsp:   req.FunCodeRsp,
		SerialNumber: serialNumber,
		Own:          client,
		IsEncrypted:  *req.IsEncrypted,
	}
	client.SetAsyncTask(&asyncTask, key)

	// 获取对应主题
	// topic, exists := global.M2A_FuncodeTopic[int(req.FunCodeReq)]
	topic := global.GetTopic(int(req.FunCodeReq))
	if topic == "" {
		global.M2A_LOG.Error(err.Error(), zap.Error(err))
		return
	}

	// 向总线发送业务信息
	packet := &global.MQPacket{
		FunCode:   req.FunCodeReq,
		SessionId: uint32(claims.SessionID),
		Data:      bytes,
	}
	go global.Publish(topic, packet)

	// 阻塞线程等待总线回复或超时
	asyncTask.Rsp = make(chan response.MQBodyRsp)
	select {
	case r := <-asyncTask.Rsp: // 总线回复
		rsp.OkWithData(r, c)
	case <-time.After(time.Second * 30): // 超时
		rsp.FailWithMessage("业务超时", c)
	}
	asyncTask.Finish()

	return
}