consumer

Documentation

Index

• type Consumer
  • func New(client *redis.Client, group, consumer string, options ...Option) *Consumer
  • func (c *Consumer) Ack(ctx context.Context, msgs ...Message) error
  • func (c *Consumer) Read(ctx context.Context) ([]Message, error)
• type Message
• type Option
  • func WithBlock(duration time.Duration) Option
  • func WithCount(cnt int64) Option
  • func WithNoAck(noAck bool) Option
  • func WithStream(stream string) Option

Types

type Consumer

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

A Consumer consumes messages from a stream.

func New

func New(client *redis.Client, group, consumer string, options ...Option) *Consumer

New creates a new consumer.

func (*Consumer) Ack

func (c *Consumer) Ack(ctx context.Context, msgs ...Message) error

Ack acknowledges the messages.

func (*Consumer) Read

func (c *Consumer) Read(ctx context.Context) ([]Message, error)

Read reads messages from the stream.

type Message

type Message struct {
	Stream string
	ID     string
	Values map[string]interface{}
}

A Message is a consumed message from a redis stream.

type Option

type Option func(*config)

An Option modifies the config.

func WithBlock

func WithBlock(duration time.Duration) Option

WithBlock sets the block field of the config.

func WithCount

func WithCount(cnt int64) Option

WithCount sets the count for the config.

func WithNoAck

func WithNoAck(noAck bool) Option

WithNoAck sets the noAck field of the config.

func WithStream

func WithStream(stream string) Option

WithStream adds a stream to the consumer.