mqx

package

v1.11.1 Latest Latest Go to latest Published: Jul 29, 2026 License: MIT Imports: 9 Imported by: 0

Details

View Source

const (
	HeaderTenantID = "x-tenant-id"
	HeaderUserID   = "x-user-id"
	HeaderBizID    = "x-biz-id"
)

This section is empty.

func ConsumeMessage(ctx context.Context, consumer mq.Consumer) (context.Context, *mq.Message, error)

ConsumeMessage 消费原始 MQ 消息，并返回从消息头恢复后的业务上下文。

func ExtractContext(ctx context.Context, msg *mq.Message) context.Context

ExtractContext 从 mq.Message.Header 中提取租户 ID、用户 ID 和业务 ID，重新注入并返回新的 Context

func InjectContext(ctx context.Context, msg *mq.Message)

InjectContext 提取 Context 中的租户 ID、用户 ID 和业务 ID，注入到 mq.Message.Header 中

func ProduceMessage(ctx context.Context, producer mq.Producer, msg *mq.Message) (*mq.ProducerResult, error)

ProduceMessage 注入上下文元数据后发送原始 MQ 消息。

type ConsumeFunc func(ctx context.Context, message *mq.Message) error

type Consumer struct {
	// contains filtered or unexported fields
}

func NewConsumer(name string, mq mq.MQ, topic string) *Consumer

func (c *Consumer) Name() string

func (c *Consumer) Start(ctx context.Context, consumeFunc ConsumeFunc) error

func (c *Consumer) Stop() error

type GeneralProducer[T any] struct {
	// contains filtered or unexported fields
}

func NewGeneralProducer[T any](q mq.MQ, topic string) (*GeneralProducer[T], error)

func (p *GeneralProducer[T]) Produce(ctx context.Context, evt T) error

type MultipleProducer[T any] struct {
	// contains filtered or unexported fields
}

MultipleProducer 管理多个 GeneralProducer

func NewMultipleProducer[T any](mq mq.MQ) *MultipleProducer[T]

NewMultipleProducer 创建一个新的 ProducerManager

func (pm *MultipleProducer[T]) AddProducer(topic string) error

AddProducer 添加一个新的 producer 监听指定的 topic

func (pm *MultipleProducer[T]) Produce(ctx context.Context, topic string, evt T) error

Produce 发送消息到指定的 topic

type Producer[T any] interface {
	Produce(ctx context.Context, evt T) error
}