class RedisQueue

Constants

PUSH_CODE
SCRIPTS

Public Class Methods

new(args = {}) click to toggle source
# File lib/redis_queue.rb, line 63
def initialize(args = {})
  args = { id: :messages, url: 'redis://localhost:6379/0' }.merge(args)
  @id             = args.delete(:id)
  @redis          = RedisConnection.new(args)
  @redis_blocking = RedisConnection.new(args)
  load_scripts
end

Public Instance Methods

clear() click to toggle source
# File lib/redis_queue.rb, line 167
def clear
  @redis.run do |redis|
    redis.del @id
    redis.del "#{@id}_in_use"
    redis.del "#{@id}_done"
    redis.del "#{@id}_failed"
  end
end
done(message) click to toggle source
# File lib/redis_queue.rb, line 86
def done(message)
  script :done, @id, message, now
end
done_list() click to toggle source
# File lib/redis_queue.rb, line 141
def done_list
  Hash[@redis.run { |redis| redis.hgetall "#{@id}_done" }]
end
done_size() click to toggle source
# File lib/redis_queue.rb, line 125
def done_size
  @redis.run { |redis| redis.hlen "#{@id}_done" }.to_i
end
fail(message) click to toggle source
# File lib/redis_queue.rb, line 82
def fail(message)
  script :fail, @id, message, now
end
failed_list() click to toggle source
# File lib/redis_queue.rb, line 145
def failed_list
  Hash[@redis.run { |redis| redis.hgetall "#{@id}_failed" }]
end
failed_size() click to toggle source
# File lib/redis_queue.rb, line 129
def failed_size
  @redis.run { |redis| redis.hlen "#{@id}_failed" }.to_i
end
forget(message) click to toggle source
# File lib/redis_queue.rb, line 98
def forget(message)
  @redis.run { |redis| redis.hdel "#{@id}_in_use", message }
end
in_use_list() click to toggle source
# File lib/redis_queue.rb, line 149
def in_use_list
  Hash[@redis.run { |redis| redis.hgetall "#{@id}_in_use" }]
end
in_use_size() click to toggle source
# File lib/redis_queue.rb, line 133
def in_use_size
  @redis.run { |redis| redis.hlen "#{@id}_in_use" }.to_i
end
list() click to toggle source
# File lib/redis_queue.rb, line 137
def list
  @redis.run { |redis| redis.lrange @id, 0, -1 }
end
pop(block: true) click to toggle source
# File lib/redis_queue.rb, line 71
def pop(block: true)
  return nonblpop unless block
  message = blpop
  @redis.run { |redis| redis.hset "#{@id}_in_use", message, now } if message
  message
end
print_contents() click to toggle source
print_stats() click to toggle source
push(message, priority = false) click to toggle source
# File lib/redis_queue.rb, line 78
def push(message, priority = false)
  script :push, @id, message, priority
end
remove(message) click to toggle source
# File lib/redis_queue.rb, line 102
def remove(message)
  @redis.run { |redis| redis.lrem @id, 0, message }
end
repush(message, priority = false) click to toggle source
# File lib/redis_queue.rb, line 94
def repush(message, priority = false)
  script :repush, @id, message, priority
end
reset(older_than: nil) click to toggle source
# File lib/redis_queue.rb, line 113
def reset(older_than: nil)
  init_from "#{@id}_in_use", older_than
end
restart() click to toggle source
# File lib/redis_queue.rb, line 117
def restart
  init_from "#{@id}_done"
end
size() click to toggle source
# File lib/redis_queue.rb, line 121
def size
  @redis.run { |redis| redis.llen @id }.to_i
end
touch(block: true) click to toggle source
# File lib/redis_queue.rb, line 106
def touch(block: true)
  return nonbltouch unless block
  message = blpop
  push(message)
  message
end
unpop(message) click to toggle source
# File lib/redis_queue.rb, line 90
def unpop(message)
  script :unpop, @id, message
end

Private Instance Methods

blpop() click to toggle source
# File lib/redis_queue.rb, line 178
def blpop
  loop do
    message = @redis_blocking.run { |redis| redis.blpop(@id) }.last
    return message unless message == ''
  end
end
init_from(key, older_than = nil) click to toggle source
# File lib/redis_queue.rb, line 193
def init_from(key, older_than = nil)
  script(:init_from, @id, key, older_than || now + 100_000)
end
load_scripts() click to toggle source
# File lib/redis_queue.rb, line 197
def load_scripts
  @scripts = {}
  @redis.run do |redis|
    SCRIPTS.each do |name, code|
      @scripts[name] = redis.script(:load, code)
    end
  end
end
nonblpop() click to toggle source
# File lib/redis_queue.rb, line 185
def nonblpop
  script :nonblpop, @id, now
end
nonbltouch() click to toggle source
# File lib/redis_queue.rb, line 189
def nonbltouch
  script :touch, @id
end
now() click to toggle source
# File lib/redis_queue.rb, line 210
def now
  (Time.now.to_f * 1000).to_i
end
script(name, *args) click to toggle source
# File lib/redis_queue.rb, line 206
def script(name, *args)
  @redis.run { |redis| redis.evalsha @scripts[name], argv: args }
end