libmq

Documentation

Index

• Variables
• type MQ
  • func NewMQ(c *conf.MQ, dbc *conf.Database, cachec *conf.Cache, app *libapp.Settings) (*MQ, func(), error)
  • func (a *MQ) Publish(ctx context.Context, topic string, payload []byte) error
  • func (a *MQ) RegisterTopic(topic TopicInterface) error
  • func (a *MQ) Start(ctx context.Context) error
  • func (a *MQ) Stop(ctx context.Context) error
• type Option
  • func WithConsumePoisoned() Option
• type Options
• type Topic
  • func NewTopic[T any](topic string, consumerFunc func(context.Context, *T) error, opts ...Option) *Topic[T]
  • func (t *Topic[T]) Consume(ctx context.Context, i []byte) error
  • func (t *Topic[T]) GetOptions() *Options
  • func (t *Topic[T]) LocalCall(ctx context.Context, i T) error
  • func (t *Topic[T]) Name() string
  • func (t *Topic[T]) Publish(ctx context.Context, i T) error
  • func (t *Topic[T]) SetMQ(mq *MQ)
• type TopicInterface

Variables

var ProviderSet = wire.NewSet(NewMQ)

Types

type MQ

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

func NewMQ

func NewMQ(c *conf.MQ, dbc *conf.Database, cachec *conf.Cache, app *libapp.Settings) (*MQ, func(), error)

func (*MQ) Publish

func (a *MQ) Publish(ctx context.Context, topic string, payload []byte) error

func (*MQ) RegisterTopic

func (a *MQ) RegisterTopic(topic TopicInterface) error

RegisterTopic register Topic to MQ If a message keep fail after retry, It will be sent to the PoisonQueue. PoisonQueue will retry messages in a very low rate.

func (*MQ) Start

func (a *MQ) Start(ctx context.Context) error

func (*MQ) Stop

func (a *MQ) Stop(ctx context.Context) error

type Option

type Option func(o *Options)

Option represents a store option function.

func WithConsumePoisoned

func WithConsumePoisoned() Option

type Options

type Options struct {
	ConsumePoisoned bool
}

type Topic

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

func NewTopic

func NewTopic[T any](topic string, consumerFunc func(context.Context, *T) error, opts ...Option) *Topic[T]

func (*Topic[T]) Consume

func (t *Topic[T]) Consume(ctx context.Context, i []byte) error

func (*Topic[T]) GetOptions

func (t *Topic[T]) GetOptions() *Options

func (*Topic[T]) LocalCall

func (t *Topic[T]) LocalCall(ctx context.Context, i T) error

func (*Topic[T]) Name

func (t *Topic[T]) Name() string

func (*Topic[T]) Publish

func (t *Topic[T]) Publish(ctx context.Context, i T) error

func (*Topic[T]) SetMQ

func (t *Topic[T]) SetMQ(mq *MQ)

type TopicInterface

type TopicInterface interface {
	Name() string
	Consume(context.Context, []byte) error
	SetMQ(*MQ)
	GetOptions() *Options
}