StartProcessLoop starts a loop that will process items from the queue it will spawn a goroutine for each worker. ctx can be used to signal shutdown Wait group is returned to allow for graceful shutdown
(ctx context.Context, workers []ProcessFunc)
| 343 | // ctx can be used to signal shutdown |
| 344 | // Wait group is returned to allow for graceful shutdown |
| 345 | func (s *RedisQueue) StartProcessLoop(ctx context.Context, workers []ProcessFunc) *sync.WaitGroup { |
| 346 | var wg sync.WaitGroup |
| 347 | for _, process := range workers { |
| 348 | wg.Add(1) |
| 349 | go func(process func(ctx context.Context, data []byte, info QueueItemInfo) error) { |
| 350 | defer wg.Done() |
| 351 | |
| 352 | exp := backoff.NewExponentialBackOff() |
| 353 | exp.MaxInterval = 30 * time.Second |
| 354 | exp.MaxElapsedTime = 2 * time.Minute |
| 355 | back := backoff.WithContext(exp, ctx) |
| 356 | for { |
| 357 | select { |
| 358 | case <-ctx.Done(): |
| 359 | return |
| 360 | default: |
| 361 | err := backoff.Retry(func() error { |
| 362 | err := s.processNextItem(ctx, process) |
| 363 | return err |
| 364 | }, back) |
| 365 | if err != nil && !errors.Is(err, context.Canceled) { |
| 366 | s.log.Error("Processing next element failed", zap.Error(err)) |
| 367 | } |
| 368 | } |
| 369 | } |
| 370 | }(process) |
| 371 | } |
| 372 | return &wg |
| 373 | } |
| 374 | |
| 375 | func (s *RedisQueue) retryItem(ctx context.Context, args packArgs, incrIteration, incrBlock bool, back backoff.BackOff) error { |
| 376 | if args.iteration >= s.Config.MaxRetries { |