func (c *Consumer) enqueue(stream string, msgs []redis.XMessage) {
	for _, m := range msgs {
		msg := &Message{
			ID:     m.ID,
			Stream: stream,
			Values: m.Values,
		}
		c.queue <- msg
	}
}
enqueue takes a slice of XMessages, creates corresponding Messages, and sends
them on the centralized channel for worker goroutines to process.
package redisqueue
import (
	"net"
	"os"
	"sync"
	"time"
	"github.com/go-redis/redis/v7"
	"github.com/pkg/errors"
)
type ConsumerFunc func(*Message) error
type registeredConsumer struct {
	fn ConsumerFunc
	id string
}
type ConsumerOptions struct {
	Name string
	GroupName string
	VisibilityTimeout time.Duration
	BlockingTimeout time.Duration
	ReclaimInterval time.Duration
	BufferSize int
	Concurrency int
	RedisClient redis.UniversalClient
	RedisOptions *RedisOptions
}
type Consumer struct {
	Errors chan error
	options   *ConsumerOptions
	redis     redis.UniversalClient
	consumers map[string]registeredConsumer
	streams   []string
	queue     chan *Message
	wg        *sync.WaitGroup
	stopReclaim chan struct{}
	stopPoll    chan struct{}
	stopWorkers chan struct{}
}
var defaultConsumerOptions = &ConsumerOptions{
	VisibilityTimeout: 60 * time.Second,
	BlockingTimeout:   5 * time.Second,
	ReclaimInterval:   1 * time.Second,
	BufferSize:        100,
	Concurrency:       10,
}
func NewConsumer() (*Consumer, error) {
	return NewConsumerWithOptions(defaultConsumerOptions)
}
func NewConsumerWithOptions(options *ConsumerOptions) (*Consumer, error) {
	hostname, _ := os.Hostname()
	if options.Name == "" {
		options.Name = hostname
	}
	if options.GroupName == "" {
		options.GroupName = "redisqueue"
	}
	if options.BlockingTimeout == 0 {
		options.BlockingTimeout = 5 * time.Second
	}
	if options.ReclaimInterval == 0 {
		options.ReclaimInterval = 1 * time.Second
	}
	var r redis.UniversalClient
	if options.RedisClient != nil {
		r = options.RedisClient
	} else {
		r = newRedisClient(options.RedisOptions)
	}
	if err := redisPreflightChecks(r); err != nil {
		return nil, err
	}
	return &Consumer{
		Errors: make(chan error),
		options:   options,
		redis:     r,
		consumers: make(map[string]registeredConsumer),
		streams:   make([]string, 0),
		queue:     make(chan *Message, options.BufferSize),
		wg:        &sync.WaitGroup{},
		stopReclaim: make(chan struct{}, 1),
		stopPoll:    make(chan struct{}, 1),
		stopWorkers: make(chan struct{}, options.Concurrency),
	}, nil
}
func (c *Consumer) RegisterWithLastID(stream string, id string, fn ConsumerFunc) {
	if len(id) == 0 {
		id = "0"
	}
	c.consumers[stream] = registeredConsumer{
		fn: fn,
		id: id,
	}
}
func (c *Consumer) Register(stream string, fn ConsumerFunc) {
	c.RegisterWithLastID(stream, "0", fn)
}
func (c *Consumer) Run() {
	if len(c.consumers) == 0 {
		c.Errors <- errors.New("at least one consumer function needs to be registered")
		return
	}
	for stream, consumer := range c.consumers {
		c.streams = append(c.streams, stream)
		err := c.redis.XGroupCreateMkStream(stream, c.options.GroupName, consumer.id).Err()
		if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
			c.Errors <- errors.Wrap(err, "error creating consumer group")
			return
		}
	}
	for i := 0; i < len(c.consumers); i++ {
		c.streams = append(c.streams, ">")
	}
	go c.reclaim()
	go c.poll()
	stop := newSignalHandler()
	go func() {
		<-stop
		c.Shutdown()
	}()
	c.wg.Add(c.options.Concurrency)
	for i := 0; i < c.options.Concurrency; i++ {
		go c.work()
	}
	c.wg.Wait()
}
func (c *Consumer) Shutdown() {
	c.stopReclaim <- struct{}{}
	if c.options.VisibilityTimeout == 0 {
		c.stopPoll <- struct{}{}
	}
}
func (c *Consumer) reclaim() {
	if c.options.VisibilityTimeout == 0 {
		return
	}
	ticker := time.NewTicker(c.options.ReclaimInterval)
	for {
		select {
		case <-c.stopReclaim:
			c.stopPoll <- struct{}{}
			return
		case <-ticker.C:
			for stream := range c.consumers {
				start := "-"
				end := "+"
				for {
					res, err := c.redis.XPendingExt(&redis.XPendingExtArgs{
						Stream: stream,
						Group:  c.options.GroupName,
						Start:  start,
						End:    end,
						Count:  int64(c.options.BufferSize - len(c.queue)),
					}).Result()
					if err != nil && err != redis.Nil {
						c.Errors <- errors.Wrap(err, "error listing pending messages")
						break
					}
					if len(res) == 0 {
						break
					}
					msgs := make([]string, 0)
					for _, r := range res {
						if r.Idle >= c.options.VisibilityTimeout {
							claimres, err := c.redis.XClaim(&redis.XClaimArgs{
								Stream:   stream,
								Group:    c.options.GroupName,
								Consumer: c.options.Name,
								MinIdle:  c.options.VisibilityTimeout,
								Messages: []string{r.ID},
							}).Result()
							if err != nil && err != redis.Nil {
								c.Errors <- errors.Wrapf(err, "error claiming %d message(s)", len(msgs))
								break
							}
							if err == redis.Nil {
								err = c.redis.XAck(stream, c.options.GroupName, r.ID).Err()
								if err != nil {
									c.Errors <- errors.Wrapf(err, "error acknowledging after failed claim for %q stream and %q message", stream, r.ID)
									continue
								}
							}
							c.enqueue(stream, claimres)
						}
					}
					newID, err := incrementMessageID(res[len(res)-1].ID)
					if err != nil {
						c.Errors <- err
						break
					}
					start = newID
				}
			}
		}
	}
}
func (c *Consumer) poll() {
	for {
		select {
		case <-c.stopPoll:
			for i := 0; i < c.options.Concurrency; i++ {
				c.stopWorkers <- struct{}{}
			}
			return
		default:
			res, err := c.redis.XReadGroup(&redis.XReadGroupArgs{
				Group:    c.options.GroupName,
				Consumer: c.options.Name,
				Streams:  c.streams,
				Count:    int64(c.options.BufferSize - len(c.queue)),
				Block:    c.options.BlockingTimeout,
			}).Result()
			if err != nil {
				if err, ok := err.(net.Error); ok && err.Timeout() {
					continue
				}
				if err == redis.Nil {
					continue
				}
				c.Errors <- errors.Wrap(err, "error reading redis stream")
				continue
			}
			for _, r := range res {
				c.enqueue(r.Stream, r.Messages)
			}
		}
	}
}