transfer_middleware/cmd/rpc/internal/queue/mq/mq.go

package mq

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/streadway/amqp"
	"time"
	"trasfer_middleware/cmd/rpc/etc"
	"trasfer_middleware/cmd/rpc/internal/logic/vo"
	"trasfer_middleware/cmd/rpc/internal/queue/mq/mqSvc"
	"trasfer_middleware/genModel"
)

type Message interface {
	MessageHandler(tag uint64, ch *amqp.Channel, msg []byte) error
}
type Market struct {
	svc *mqSvc.ServiceContext
	ctx context.Context
}
type ZLTX struct {
	svc *mqSvc.ServiceContext
	ctx context.Context
}

type RS struct {
	svc *mqSvc.ServiceContext
	ctx context.Context
}

func NewMarket(svc *mqSvc.ServiceContext, ctx context.Context) *Market {
	return &Market{
		svc: svc,
		ctx: ctx,
	}
}

func NewZLTX(svc *mqSvc.ServiceContext, ctx context.Context) *ZLTX {
	return &ZLTX{
		svc: svc,
		ctx: ctx,
	}
}

func NewRS(svc *mqSvc.ServiceContext, ctx context.Context) *RS {
	return &RS{
		svc: svc,
		ctx: ctx,
	}
}

func (m *Market) MessageHandler(tag uint64, ch *amqp.Channel, msg []byte) error {
	var market = &genModel.ServerMiddleMarketLogs{}
	json.Unmarshal(msg, market)
	market.CreateTime = time.Now()
	market.Status = vo.MARKET_LOG_STATU_DEFAULT
	_, err := m.svc.DbWrite.MarketLogs.Insert(m.ctx, market)
	if err != nil {
		return fmt.Errorf("market数据保存失败：%s,原因：%s", msg, err)
	}
	return nil
}

func (m *ZLTX) MessageHandler(tag uint64, ch *amqp.Channel, msg []byte) error {
	var zltx = &genModel.ServerMiddleZltxLogs{}
	json.Unmarshal(msg, zltx)
	zltx.CreateTime = time.Now()
	zltx.Status = vo.ZLTX_LOG_STATU_DEFAULT
	_, err := m.svc.DbWrite.ZLTXLogs.Insert(m.ctx, zltx)
	if err != nil {
		return fmt.Errorf("zltx数据保存失败：%s,原因：%s", msg, err)
	}
	return nil
}

func (m *RS) MessageHandler(tag uint64, ch *amqp.Channel, msg []byte) error {
	var rs = &genModel.ServerMiddleRsLogs{}
	json.Unmarshal(msg, rs)
	rs.CreateTime = time.Now()
	rs.Status = vo.RS_LOG_STATU_DEFAULT
	_, err := m.svc.DbWrite.RSLogs.Insert(m.ctx, rs)
	if err != nil {
		return fmt.Errorf("rs数据保存失败：%s,原因：%s", msg, err)
	}
	return nil
}

func AllHandle(c *etc.RockerMqConfig, svc *mqSvc.ServiceContext, ctx context.Context) map[string]Message {
	result := make(map[string]Message)
	result[c.TopicPrefix+c.Topic.Market.Name] = NewMarket(svc, ctx)
	result[c.TopicPrefix+c.Topic.ZLTX.Name] = NewZLTX(svc, ctx)
	result[c.TopicPrefix+c.Topic.RS.Name] = NewRS(svc, ctx)
	return result
}