stream

package

v0.34.0 Latest Latest Go to latest Published: Oct 1, 2024 License: AGPL-3.0, AGPL-3.0-only Imports: 4 Imported by: 0

Details

This section is empty.

This section is empty.

This section is empty.

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

A Consumer consumes messages from a stream.

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

New creates a new consumer.

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

Ack acknowledges the messages.

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

Read reads messages from the stream.

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

A Message is a consumed message from a redis stream.

type Option func(*config)

An Option modifies the config.

func WithBlock(duration time.Duration) Option

WithBlock sets the block field of the config.

func WithCount(cnt int64) Option

WithCount sets the count for the config.

func WithNoAck(noAck bool) Option

WithNoAck sets the noAck field of the config.

func WithStream(stream string) Option

WithStream adds a stream to the consumer.