Function popTask

pkg/redis/redis.go:556–626

popTask calls f on the most recent task in the queue, for which timestamp is in range [0, time.Now()] or blocks until such is available or context is done. group is the consumer group name. consumer is the consumer group ID. ReadyTaskKey(k) is the keys to pop from. Pipeline is executed even if f ret

(
	ctx context.Context,
	r redis.Cmdable,
	group, consumer string,
	f func(p redis.Pipeliner, payload string, startAt time.Time) error,
	k string,
	blockLimit time.Duration,
)

// Pipeline is executed even if f returns an error.
// Tasks are acked only if f returns without error.
func popTask(
	ctx context.Context,
	r redis.Cmdable,
	group, consumer string,
	f func(p redis.Pipeliner, payload string, startAt time.Time) error,
	k string,
	blockLimit time.Duration,
) (err error) {
	readyStream := ReadyTaskKey(k)

	processMessage := func(message redis.XMessage) error {
		fields := make(map[string]string, len(message.Values))
		for k, v := range message.Values {
			val, ok := v.(string)
			if !ok {
				panic(fmt.Sprintf("invalid field type %T", v))
			}
			fields[k] = val
		}

		var startAt time.Time
		if s, ok := fields[startAtKey]; ok {
			startAt, err = parseTime(s)
			if err != nil {
				return errInvalidKeyValueType.WithAttributes("key", startAtKey).WithCause(err)
			}
		}

		p := r.Pipeline()
		defer func() {
			// Ensure pipeline is executed even if f fails.
			_, pErr := p.Exec(ctx)
			if err == nil && pErr != nil {
				err = ConvertError(pErr)
			}
		}()

		if err = f(p, fields[payloadKey], startAt); err != nil {
			return err
		}

		p.XAck(ctx, readyStream, group, message.ID)
		p.XDel(ctx, readyStream, message.ID)

		return nil
	}

	var xs []redis.XStream
	for len(xs) == 0 {
		xs, err = r.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    group,
			Consumer: consumer,
			Streams:  []string{readyStream, ">"},
			Count:    1,
			Block:    blockLimit,
		}).Result()
		if err != nil && !errors.Is(err, redis.Nil) {
			return ConvertError(err)