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
				}
			}
		}
	}
}
reclaim runs in a separate goroutine and checks the list of pending messages
in every stream. For every message, if it's been idle for longer than the
VisibilityTimeout, it will attempt to claim that message for this consumer.
If VisibilityTimeout is 0, this function returns early and no messages are
reclaimed. It checks the list of pending messages according to
ReclaimInterval.
package redisqueue
import (
	"net"
	"os"
	"sync"
	"time"
	"github.com/go-redis/redis"
	"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.Client
	RedisOptions *RedisOptions
}
type Consumer struct {
	Errors chan error
	options   *ConsumerOptions
	redis     *redis.Client
	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.Client
	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{}{}
	}
}