mq

Documentation

Index

• type Consumer
• type Queue
• type StreamQueue
  • func NewStreamQueue(client *redis.Client) *StreamQueue
  • func (h *StreamQueue) Consume(ctx context.Context, topic, group, consumer string, batchSize int64, ...) error
  • func (h *StreamQueue) Publish(ctx context.Context, topic string, msg any, maxLen int64) (id string, err error)

Types

type Consumer

type Consumer func(ctx context.Context, client *redis.Client, topic, group string, id string, value any) error

Consumer will be called while received messages

type Queue

type Queue interface {
	// Publish publishes a message to the specified topic
	Publish(ctx context.Context, topic string, msg any, maxLen int64) (id string, err error)
	// Consume register a consumer with callback
	Consume(ctx context.Context, topic, group, consumer string, batchSize int64, cb Consumer) error
}

Queue define a set of methods that message queue handler should implement

type StreamQueue

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

StreamQueue implement Queue interface by Redis Stream

func NewStreamQueue

func NewStreamQueue(client *redis.Client) *StreamQueue

func (*StreamQueue) Consume

func (h *StreamQueue) Consume(ctx context.Context, topic, group, consumer string, batchSize int64, cb Consumer) error

func (*StreamQueue) Publish

func (h *StreamQueue) Publish(ctx context.Context, topic string, msg any, maxLen int64) (id string, err error)