(ctx context.Context, args packArgs)
| 222 | } |
| 223 | |
| 224 | func (s *RedisQueue) pushToQueue(ctx context.Context, args packArgs) error { |
| 225 | processable, unprocessable, err := s.queuedItems(ctx) |
| 226 | if err != nil { |
| 227 | s.log.Warn("failed to get queued items", zap.Error(err)) |
| 228 | return err |
| 229 | } |
| 230 | |
| 231 | nextBlock := atomic.LoadUint64(s.currentBlock) + 1 |
| 232 | |
| 233 | canProcessItem := args.minTargetBlock <= nextBlock |
| 234 | |
| 235 | if canProcessItem { |
| 236 | threshold := s.Config.MaxQueuedProcessableItemsLowPrio |
| 237 | if args.highPriority { |
| 238 | threshold = s.Config.MaxQueuedProcessableItemsHighPrio |
| 239 | } |
| 240 | if processable >= threshold { |
| 241 | s.log.Error("too many queued processable items in the queue", zap.Uint64("max_processable_items", threshold)) |
| 242 | return ErrQueueFull |
| 243 | } |
| 244 | } else { |
| 245 | threshold := s.Config.MaxQueuedUnprocessableItemsLowPrio |
| 246 | if args.highPriority { |
| 247 | threshold = s.Config.MaxQueuedUnprocessableItemsHighPrio |
| 248 | } |
| 249 | if unprocessable >= threshold { |
| 250 | s.log.Error("too many queued unprocessable items in the queue", zap.Uint64("max_unprocessable_items", threshold)) |
| 251 | return ErrQueueFull |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | score, redisData := packData(args) |
| 256 | err = s.red.ZAdd(ctx, s.queueName, redis.Z{Score: score, Member: redisData}).Err() |
| 257 | if err != nil { |
| 258 | s.log.Debug("failed to push to queue", zap.Error(err)) |
| 259 | } |
| 260 | return err |
| 261 | } |
| 262 | |
| 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 |
no test coverage detected