popFromQueue pops an item from the queue it will block for up to 1 second waiting for an item if a queue is empty
(ctx context.Context)
| 263 | // popFromQueue pops an item from the queue |
| 264 | // it will block for up to 1 second waiting for an item if a queue is empty |
| 265 | func (s *RedisQueue) popFromQueue(ctx context.Context) (packArgs, error) { |
| 266 | // 1 second is minimal value for a timeout |
| 267 | // we will block for up to 1 second waiting for an item |
| 268 | value, err := s.red.BZPopMin(ctx, time.Second, s.queueName).Result() |
| 269 | if err != nil { |
| 270 | if errors.Is(err, redis.Nil) { |
| 271 | return packArgs{}, err |
| 272 | } |
| 273 | s.log.Error("failed to pop from queue", zap.Error(err)) |
| 274 | return packArgs{}, err |
| 275 | } |
| 276 | |
| 277 | redisData, ok := value.Member.(string) |
| 278 | if !ok { |
| 279 | s.log.Error("failed to pop from queue, invalid data type") |
| 280 | return packArgs{}, err |
| 281 | } |
| 282 | |
| 283 | args, err := unpackData(value.Score, []byte(redisData)) |
| 284 | if err != nil { |
| 285 | s.log.Error("failed to unpack data", zap.Error(err)) |
| 286 | return packArgs{}, err |
| 287 | } |
| 288 | return args, nil |
| 289 | } |
| 290 | |
| 291 | func (s *RedisQueue) processNextItem(ctx context.Context, process ProcessFunc) error { |
| 292 | // we use this backoff for requeuing items because It's important to not lose items |
no test coverage detected