Class: Specbandit::RedisQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/specbandit/redis_queue.rb

Constant Summary collapse

MAX_BACKOFF_SECONDS =

Cap the exponential backoff so a real outage degrades/fails within a bounded window instead of sleeping for minutes on the last attempts.

10

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis_url: Specbandit.configuration.redis_url, connect_timeout: Specbandit.configuration.redis_connect_timeout, read_timeout: Specbandit.configuration.redis_timeout, write_timeout: Specbandit.configuration.redis_timeout, reconnect_attempts: Specbandit.configuration.redis_reconnect_attempts, max_attempts: Specbandit.configuration.redis_max_attempts) ⇒ RedisQueue

Returns a new instance of RedisQueue.



13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
# File 'lib/specbandit/redis_queue.rb', line 13

def initialize(
  redis_url: Specbandit.configuration.redis_url,
  connect_timeout: Specbandit.configuration.redis_connect_timeout,
  read_timeout: Specbandit.configuration.redis_timeout,
  write_timeout: Specbandit.configuration.redis_timeout,
  reconnect_attempts: Specbandit.configuration.redis_reconnect_attempts,
  max_attempts: Specbandit.configuration.redis_max_attempts
)
  @max_attempts = max_attempts
  @redis = Redis.new(
    url: redis_url,
    connect_timeout: connect_timeout,
    read_timeout: read_timeout,
    write_timeout: write_timeout,
    reconnect_attempts: reconnect_attempts
  )
end

Instance Attribute Details

#redis ⇒ Object (readonly)

Returns the value of attribute redis.



7
8
9
 
# File 'lib/specbandit/redis_queue.rb', line 7

def redis
  @redis
end

Instance Method Details

#close ⇒ Object 



91
92
93
 
# File 'lib/specbandit/redis_queue.rb', line 91

def close
  redis.close
end

#length(key) ⇒ Object

Returns the current length of the queue.



60
61
62
 
# File 'lib/specbandit/redis_queue.rb', line 60

def length(key)
  with_retries { redis.llen(key) }
end

#mark_published(key, ttl: nil) ⇒ Object

Mark a queue key as "published" by setting a companion marker key.

The marker is a separate string key (<key>:published) that survives the queue list being fully drained. It is the source of truth for "was work ever pushed for this key?" -- Redis auto-deletes empty lists, so the list alone cannot distinguish "never pushed" from "drained".



70
71
72
73
74
75
76
77
78
 
# File 'lib/specbandit/redis_queue.rb', line 70

def mark_published(key, ttl: nil)
  with_retries do
    if ttl
      redis.set(published_marker(key), '1', ex: ttl)
    else
      redis.set(published_marker(key), '1')
    end
  end
end

#published?(key) ⇒ Boolean

Whether a queue key has been published (its marker exists).

Returns:

  • (Boolean) 


81
82
83
 
# File 'lib/specbandit/redis_queue.rb', line 81

def published?(key)
  with_retries { redis.exists?(published_marker(key)) }
end

#push(key, files, ttl: nil) ⇒ Object

Push file paths onto the queue and set an expiry on the key. Returns the new length of the list.



33
34
35
36
37
38
39
40
41
 
# File 'lib/specbandit/redis_queue.rb', line 33

def push(key, files, ttl: nil)
  return 0 if files.empty?

  with_retries do
    count = redis.rpush(key, files)
    redis.expire(key, ttl) if ttl
    count
  end
end

#read_all(key) ⇒ Object

Read all file paths from the list non-destructively. Returns an array of file paths (empty array when key doesn't exist).



87
88
89
 
# File 'lib/specbandit/redis_queue.rb', line 87

def read_all(key)
  with_retries { redis.lrange(key, 0, -1) }
end

#steal(key, count) ⇒ Object

Atomically steal up to count file paths from the queue. Returns an array of file paths (empty array when exhausted).

Uses LPOP with count argument (Redis 6.2+).



47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/specbandit/redis_queue.rb', line 47

def steal(key, count)
  result = with_retries { redis.lpop(key, count) }

  # LPOP returns nil when the key doesn't exist or list is empty,
  # and returns a single string (not array) when count is 1 on some versions.
  case result
  when nil then []
  when String then [result]
  else Array(result)
  end
end