Function dispatchTask

pkg/redis/redis.go:484–548

dispatchTask dispatches tasks for the callers of popTask. At least one dispatcher is required in order to use popTask. The tasks are moved from InputTaskKey(k) to WaitingTaskKey(k). Once the task should be dispatched, it is moved form WaitingTaskKey(k) to ReadyTaskKey(k). group is the consumer group

(
	ctx context.Context,
	r redis.Cmdable,
	group, consumer string,
	maxLen int64,
	k string,
	blockLimit time.Duration,
)

// maxLen represents the maximum size of the streams used for dispatching.
// minIdleTime is used for automatic task reclaiming. Only tasks older than minIdleTime will be redispatched from the input stream to the waiting stream.
func dispatchTask(
	ctx context.Context,
	r redis.Cmdable,
	group, consumer string,
	maxLen int64,
	k string,
	blockLimit time.Duration,
) error {
	var (
		readyStream   = ReadyTaskKey(k)
		inputStream   = InputTaskKey(k)
		waitingStream = WaitingTaskKey(k)
	)
	for {
		ret, err := dispatchTaskScript.Run(
			ctx,
			r,
			[]string{readyStream, inputStream, waitingStream},
			group,
			consumer,
			time.Now().UnixNano(),
			maxLen,
		).Result()
		if err != nil && !errors.Is(err, redis.Nil) {
			return ConvertError(err)
		}

		block := blockLimit
		if ret != nil {
			s, ok := ret.(string)
			if !ok {
				return errInvalidKeyValueType.WithAttributes("key", nextAtKey).WithCause(err)
			}
			nextAt, err := parseTime(s)
			if err != nil {
				return errInvalidKeyValueType.WithAttributes("key", nextAtKey).WithCause(err)
			}
			if nextAt.IsZero() {
				block = -1
			} else {
				now := time.Now()
				if nextAt.Before(now) {
					continue
				}
				// If we have a task that we may dispatch into the future, we will block the
				// input stream only for the duration between the current time and that future
				// time.
				if d := nextAt.Sub(now); block == 0 || d < block {
					block = d
				}
			}
		}

		_, err = r.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    group,
			Consumer: consumer,
			Streams:  []string{inputStream, ">"},
			Count:    1,