Class: Sidekiq::Tasks::Storage::Redis

Inherits:
Base
  • Object
show all
Defined in:
lib/sidekiq/tasks/storage/redis.rb

Constant Summary collapse

JID_PREFIX = 
"task".freeze

Constants inherited from Base

Base::ERROR_MESSAGE_MAX_LENGTH

Instance Attribute Summary

Attributes inherited from Base

#history_limit, #task_name

Instance Method Summary collapse

Methods inherited from Base

#initialize

Constructor Details

This class inherits a constructor from Sidekiq::Tasks::Storage::Base

Instance Method Details

#history ⇒ Object 



18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 18

def history
  raw_entries = Sidekiq.redis { |conn| conn.lrange(history_key, 0, -1) }

  return [] unless raw_entries

  raw_entries.map do |raw|
    entry = Sidekiq.load_json(raw)
    ["enqueued_at", "executed_at", "finished_at"].each do |key|
      entry[key] = Time.at(entry[key]) if entry[key]
    end
    entry
  end
end

#history_key ⇒ Object 



10
11
12
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 10

def history_key
  "#{jid_key}:history"
end

#jid_key ⇒ Object 



6
7
8
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 6

def jid_key
  "#{JID_PREFIX}:#{task_name}"
end

#last_enqueue_at ⇒ Object 



14
15
16
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 14

def last_enqueue_at
  stored_time("last_enqueue_at")
end

#store_enqueue(jid, args, user: nil) ⇒ Object 



41
42
43
44
45
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 41

def store_enqueue(jid, args, user: nil)
  time = Time.now.to_f
  store_time(time, "last_enqueue_at")
  store_history(jid, args, time, user: user)
end

#store_execution(jid, time_key) ⇒ Object 



47
48
49
50
51
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 47

def store_execution(jid, time_key)
  update_history_entry(jid) do |entry|
    entry.merge(time_key => Time.now.to_f)
  end
end

#store_execution_error(jid, error) ⇒ Object 



53
54
55
56
57
58
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 53

def store_execution_error(jid, error)
  update_history_entry(jid) do |entry|
    error_message = truncate_message("#{error.class}: #{error.message}", ERROR_MESSAGE_MAX_LENGTH)
    entry.merge("error" => error_message)
  end
end

#store_history(jid, task_args, time, user: nil) ⇒ Object 



32
33
34
35
36
37
38
39
 
# File 'lib/sidekiq/tasks/storage/redis.rb', line 32

def store_history(jid, task_args, time, user: nil)
  Sidekiq.redis do |conn|
    task_trace = {jid: jid, task_name: task_name, args: task_args, enqueued_at: time.to_f}
    task_trace[:user] = user if user
    conn.lpush(history_key, Sidekiq.dump_json(task_trace))
    conn.ltrim(history_key, 0, history_limit - 1)
  end
end