forked from adamwiggins/stalker
-
Notifications
You must be signed in to change notification settings - Fork 1
/
squeese.rb
105 lines (81 loc) · 1.9 KB
/
squeese.rb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
require 'active_support' # workaround for bug with directly requiring aws
require 'aws'
require 'json'
require 'uri'
module Squeese
extend self
def purge
queue.delete
end
def enqueue(job, args={})
queue.send_message [ job, args ].to_json
end
def job(j, &block)
@@handlers ||= {}
@@handlers[j] = block
end
class NoJobsDefined < RuntimeError; end
class NoSuchJob < RuntimeError; end
def work
raise NoJobsDefined unless defined?(@@handlers)
# this makes more sense when we get support for working a subset
# of the available jobs.
jobs = all_jobs
jobs.each do |job|
raise(NoSuchJob, job) unless @@handlers[job]
end
log "Working #{jobs.size} jobs :: [ #{jobs.join(' ')} ]"
loop do
work_one_job
end
end
def work_one_job
msg = queue.pop
# don't be CPU greedy on a quiet queue
unless msg
sleep 2
return
end
name, args = JSON.parse msg.body
args = Hash.new {|h,k| h[k.to_s] if h.keys.include? k.to_s}.merge(args)
log_job(name, args)
handler = @@handlers[name]
raise(NoSuchJob, name) unless handler
handler.call(args)
rescue => e
log exception_message(e)
log "Deleted failed job." if msg
end
def log_job(name, args)
args_flat = args.inject("") do |accum, (key,value)|
accum += "#{key}=#{value} "
end
log sprintf("%-15s :: #{args_flat}", name)
end
def log(msg)
puts "[Squeese][#{Time.now}] #{msg}"
end
def sqs
@sqs ||= Aws::Sqs.new(ENV['AWS_ACCESS_KEY_ID'], ENV['AWS_SECRET_ACCESS_KEY'], :logger => Logger.new(nil))
end
def queue_name=(val)
@@queue_name = val
end
def queue_name
@@queue_name ||= "squeese"
end
def queue
sqs.queue(queue_name, true)
end
def exception_message(e)
msg = [ "Exception #{e.class} -> #{e.message}" ]
base = File.expand_path(Dir.pwd) + '/'
e.backtrace.each do |t|
msg << " #{File.expand_path(t).gsub(/#{base}/, '')}"
end
msg.join("\n")
end
def all_jobs
@@handlers.keys
end
end