Class: SweatShop::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/sweat_shop/worker.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.after_task(&block) ⇒ Object 



138
139
140
141
142
143
144
 
# File 'lib/sweat_shop/worker.rb', line 138

def self.after_task(&block)
  if block
    @after_task = block
  else
    @after_task
  end
end

.before_task(&block) ⇒ Object 



130
131
132
133
134
135
136
 
# File 'lib/sweat_shop/worker.rb', line 130

def self.before_task(&block)
  if block
    @before_task = block
  else
    @before_task
  end
end

.call_after_task(task) ⇒ Object 



120
121
122
123
 
# File 'lib/sweat_shop/worker.rb', line 120

def self.call_after_task(task)
  superclass.call_after_task(task) if superclass.respond_to?(:call_after_task)
  after_task.call(task) if after_task
end

.call_before_task(task) ⇒ Object 



115
116
117
118
 
# File 'lib/sweat_shop/worker.rb', line 115

def self.call_before_task(task)
  superclass.call_before_task(task) if superclass.respond_to?(:call_before_task)
  before_task.call(task) if before_task
end

.call_exception_handler(exception) ⇒ Object 



125
126
127
128
 
# File 'lib/sweat_shop/worker.rb', line 125

def self.call_exception_handler(exception)
  superclass.call_exception_handler(exception) if superclass.respond_to?(:call_exception_handler)
  on_exception.call(exception) if on_exception
end

.config ⇒ Object 



36
37
38
 
# File 'lib/sweat_shop/worker.rb', line 36

def self.config
  SweatShop.config
end

.confirm ⇒ Object 



65
66
67
 
# File 'lib/sweat_shop/worker.rb', line 65

def self.confirm
  queue.confirm(queue_name)
end

.delete_queue ⇒ Object 



44
45
46
 
# File 'lib/sweat_shop/worker.rb', line 44

def self.delete_queue
  queue.delete(queue_name)
end

.dequeue ⇒ Object 



57
58
59
60
61
62
63
 
# File 'lib/sweat_shop/worker.rb', line 57

def self.dequeue
  task = queue.dequeue(queue_name)
  unless task.nil?
    log("Receive message: #{task.inspect} from #{queue_name} by #{self.inspect}")
  end
  task
end

.do_task(task) ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/sweat_shop/worker.rb', line 81

def self.do_task(task)
  begin
    call_before_task(task)

    queued_at = task[:queued_at] ? "(queued #{Time.at(task[:queued_at]).strftime('%Y/%m/%d %H:%M:%S')})" : ''
    log("Dequeuing #{queue_name}::#{task[:method]} #{queued_at}")
    task[:result] = instance.send(task[:method], *task[:args])

    call_after_task(task)
    confirm
  rescue SystemExit
    exit
  rescue Exception => e
    log("Caught Exception: #{e.message}, \n#{e.backtrace.join("\n")}")
    call_exception_handler(e)
  end
end

.do_tasks ⇒ Object 



75
76
77
78
79
 
# File 'lib/sweat_shop/worker.rb', line 75

def self.do_tasks
  while task = dequeue
    do_task(task)
  end
end

.enqueue(task) ⇒ Object 



52
53
54
55
 
# File 'lib/sweat_shop/worker.rb', line 52

def self.enqueue(task)
  log("Send message: #{task.inspect} to #{queue_name} by #{self.inspect}")
  queue.enqueue(queue_name, task)
end

.inherited(subclass) ⇒ Object 



5
6
7
 
# File 'lib/sweat_shop/worker.rb', line 5

def self.inherited(subclass)
  self.workers << subclass
end

.instance ⇒ Object 



32
33
34
 
# File 'lib/sweat_shop/worker.rb', line 32

def self.instance
  @instance ||= new
end

.log(msg) ⇒ Object 



111
112
113
 
# File 'lib/sweat_shop/worker.rb', line 111

def self.log(msg)
  SweatShop.log(msg)
end

.method_missing(method, *args, &block) ⇒ Object 



9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/sweat_shop/worker.rb', line 9

def self.method_missing(method, *args, &block)
  if method.to_s =~ /^async_(.*)/
    method        = $1
    expected_args = instance.method(method).arity
    if expected_args != args.size
      raise ArgumentError.new("#{method} expects #{expected_args} arguments")
    end
    return instance.send(method, *args) unless config['enable']

    uid  = ::Digest::MD5.hexdigest("#{name}:#{method}:#{args}:#{Time.now.to_f}")
    task = {:args => args, :method => method, :uid => uid, :queued_at => Time.now.to_i}

    log("Putting #{uid} on #{queue_name}")
    enqueue(task)

    uid
  elsif instance.respond_to?(method)
    instance.send(method, *args)
  else
    super
  end
end

.on_exception(&block) ⇒ Object 



146
147
148
149
150
151
152
 
# File 'lib/sweat_shop/worker.rb', line 146

def self.on_exception(&block)
  if block
    @on_exception = block
  else
    @on_exception
  end
end

.queue ⇒ Object 



99
100
101
 
# File 'lib/sweat_shop/worker.rb', line 99

def self.queue
  SweatShop.queue(queue_group.to_s)
end

.queue_group(group = nil) ⇒ Object 



162
163
164
 
# File 'lib/sweat_shop/worker.rb', line 162

def self.queue_group(group=nil)
  group ? meta_def(:_queue_group){ group } : _queue_group
end

.queue_name ⇒ Object 



40
41
42
 
# File 'lib/sweat_shop/worker.rb', line 40

def self.queue_name
  @queue_name ||= self.to_s
end

.queue_size ⇒ Object 



48
49
50
 
# File 'lib/sweat_shop/worker.rb', line 48

def self.queue_size
  queue.queue_size(queue_name)
end

.stop ⇒ Object 



154
155
156
 
# File 'lib/sweat_shop/worker.rb', line 154

def self.stop
  instance.stop
end

.subscribe ⇒ Object 



69
70
71
72
73
 
# File 'lib/sweat_shop/worker.rb', line 69

def self.subscribe
  queue.subscribe(queue_name) do |task|
    do_task(task)
  end
end

.workers ⇒ Object 



103
104
105
 
# File 'lib/sweat_shop/worker.rb', line 103

def self.workers
  SweatShop.workers
end

Instance Method Details

#stop ⇒ Object

called before we exit -- subclass can implement this method



159
 
# File 'lib/sweat_shop/worker.rb', line 159

def stop; end