Class: Waitmate::Store::Redis

Inherits:
Object
  • Object
show all
Defined in:
lib/waitmate/store/redis.rb

Overview

Redis-backed store contract implementation. All operations that touch capacity or ordering run inside atomic Lua scripts so concurrent callers cannot double-count or lose slots.

Per-queue data model:

waitmate:queue:{q}          - sorted set keyed by enqueue timestamp (FIFO)
waitmate:waiting_expiry:{q} - sorted set keyed by expiry timestamp
waitmate:active:{q}         - sorted set keyed by expiry timestamp

Constant Summary collapse

KEY_PREFIX = 
"waitmate"
LUA_SCRIPTS = 
{
  enqueue: <<~LUA,

  position: <<~LUA,

  active_count: <<~LUA,

  admit: <<~LUA,

  heartbeat: <<~LUA,

  expire_stale: <<~LUA
    local queue_key = KEYS[1]
    local expiry_key = KEYS[2]
    local active_key = KEYS[3]
    local now = tonumber(ARGV[1])

    local expired_waiting = redis.call('ZRANGEBYSCORE', expiry_key, '-inf', now)
    for i = 1, #expired_waiting do
      redis.call('ZREM', queue_key, expired_waiting[i])
      redis.call('ZREM', expiry_key, expired_waiting[i])
    end

    local active_removed = redis.call('ZREMRANGEBYSCORE', active_key, '-inf', now)

    return {#expired_waiting, active_removed}
  LUA
}.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis: nil, config: Waitmate.configuration) ⇒ Redis

Returns a new instance of Redis.



161
162
163
164
165
166
167
168
169
 
# File 'lib/waitmate/store/redis.rb', line 161

def initialize(redis: nil, config: Waitmate.configuration)
  unless defined?(::Redis)
    raise ConfigurationError,
      "Redis gem is not available. Add `gem 'redis'` to your Gemfile to use the Redis adapter."
  end

  @config = config
  @redis = redis || ::Redis.new
end

Instance Attribute Details

#redis ⇒ Object (readonly)

Returns the value of attribute redis.



171
172
173
 
# File 'lib/waitmate/store/redis.rb', line 171

def redis
  @redis
end

Instance Method Details

#active_count(queue_name) ⇒ Object 



195
196
197
198
199
200
201
 
# File 'lib/waitmate/store/redis.rb', line 195

def active_count(queue_name)
  eval_script(
    :active_count,
    keys: [active_key(queue_name)],
    argv: [current_timestamp]
  )
end

#admit(queue_name, max_concurrent, count: max_concurrent) ⇒ Object 



203
204
205
206
207
208
209
 
# File 'lib/waitmate/store/redis.rb', line 203

def admit(queue_name, max_concurrent, count: max_concurrent)
  eval_script(
    :admit,
    keys: [queue_key(queue_name), expiry_key(queue_name), active_key(queue_name)],
    argv: [current_timestamp, @config.queue_ttl, max_concurrent, count]
  )
end

#enqueue(queue_name, identity, ttl: nil) ⇒ Object 



173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/waitmate/store/redis.rb', line 173

def enqueue(queue_name, identity, ttl: nil)
  ttl ||= @config.queue_ttl
  now = current_timestamp
  expiry = now + ttl

  eval_script(
    :enqueue,
    keys: [queue_key(queue_name), expiry_key(queue_name), active_key(queue_name)],
    argv: [identity.to_s, now, expiry]
  )
end

#expire_stale(queue_name) ⇒ Object 



228
229
230
231
232
233
234
235
236
 
# File 'lib/waitmate/store/redis.rb', line 228

def expire_stale(queue_name)
  waiting, active = eval_script(
    :expire_stale,
    keys: [queue_key(queue_name), expiry_key(queue_name), active_key(queue_name)],
    argv: [current_timestamp]
  )

  {waiting: waiting, active: active}
end

#heartbeat(queue_name, identity, ttl: nil) ⇒ Object 



218
219
220
221
222
223
224
225
226
 
# File 'lib/waitmate/store/redis.rb', line 218

def heartbeat(queue_name, identity, ttl: nil)
  ttl ||= @config.queue_ttl
  result = eval_script(
    :heartbeat,
    keys: [queue_key(queue_name), expiry_key(queue_name), active_key(queue_name)],
    argv: [identity.to_s, current_timestamp, ttl]
  )
  result == 1
end

#position(queue_name, identity) ⇒ Object 



185
186
187
188
189
190
191
192
193
 
# File 'lib/waitmate/store/redis.rb', line 185

def position(queue_name, identity)
  rank = eval_script(
    :position,
    keys: [queue_key(queue_name), expiry_key(queue_name), active_key(queue_name)],
    argv: [identity.to_s, current_timestamp]
  )

  (rank == -1) ? nil : rank
end

#release(queue_name, identity) ⇒ Object 



211
212
213
214
215
216
 
# File 'lib/waitmate/store/redis.rb', line 211

def release(queue_name, identity)
  result = with_redis do |redis|
    redis.zrem(active_key(queue_name), identity.to_s)
  end
  !!result
end