Module: Faktory::Job::ClassMethods
- Defined in:
- lib/faktory/testing.rb,
lib/faktory/job.rb
Overview
The Faktory testing infrastructure overrides perform_async so that it does not actually touch the network. Instead it stores the asynchronous jobs in a per-class array so that their presence/absence can be asserted by your tests.
This is similar to ActionMailer’s :test delivery_method and its ActionMailer::Base.deliveries array.
Example:
require 'faktory/testing'
assert_equal 0, HardWorker.jobs.size
HardWorker.perform_async(:something)
assert_equal 1, HardWorker.jobs.size
assert_equal :something, HardWorker.jobs[0]['args'][0]
assert_equal 0, Faktory::Extensions::DelayedMailer.jobs.size
MyMailer.delay.send_welcome_email('[email protected]')
assert_equal 1, Faktory::Extensions::DelayedMailer.jobs.size
You can also clear and drain all workers’ jobs:
assert_equal 0, Faktory::Extensions::DelayedMailer.jobs.size
assert_equal 0, Faktory::Extensions::DelayedModel.jobs.size
MyMailer.delay.send_welcome_email('[email protected]')
MyModel.delay.do_something_hard
assert_equal 1, Faktory::Extensions::DelayedMailer.jobs.size
assert_equal 1, Faktory::Extensions::DelayedModel.jobs.size
Faktory::Worker.clear_all # or .drain_all
assert_equal 0, Faktory::Extensions::DelayedMailer.jobs.size
assert_equal 0, Faktory::Extensions::DelayedModel.jobs.size
This can be useful to make sure jobs don’t linger between tests:
RSpec.configure do |config|
config.before(:each) do
Faktory::Worker.clear_all
end
end
or for acceptance testing, i.e. with cucumber:
AfterStep do
Faktory::Worker.drain_all
end
When I sign up as "[email protected]"
Then I should receive a welcome email to "[email protected]"
Instance Method Summary collapse
-
#clear ⇒ Object
Clear all jobs for this worker.
-
#client_push(item) ⇒ Object
:nodoc:.
-
#drain ⇒ Object
Drain and run all jobs for this worker.
- #execute_job(worker, args) ⇒ Object
- #faktory_class_attribute(*attrs) ⇒ Object
-
#faktory_options(opts = {}) ⇒ Object
Allows customization of Faktory features for this type of Job.
-
#get_faktory_options ⇒ Object
:nodoc:.
-
#jobs ⇒ Object
Jobs queued for this worker.
- #perform_async(*args) ⇒ Object
-
#perform_in(interval, *args) ⇒ Object
(also: #perform_at)
interval
must be a timestamp, numeric or something that acts numeric (like an activesupport time interval). -
#perform_one ⇒ Object
Pop out a single job and perform it.
- #process_job(job) ⇒ Object
-
#queue ⇒ Object
Queue for this worker.
- #set(options) ⇒ Object
Instance Method Details
#clear ⇒ Object
Clear all jobs for this worker
274 275 276 |
# File 'lib/faktory/testing.rb', line 274 def clear Queues.clear_for(queue, self.to_s) end |
#client_push(item) ⇒ Object
:nodoc:
105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
# File 'lib/faktory/job.rb', line 105 def client_push(item) # :nodoc: pool = Thread.current[:faktory_via_pool] || ['pool'.freeze] || Faktory.server_pool item = .merge(item) # stringify item.keys.each do |key| item[key.to_s] = item.delete(key) end item["jid"] ||= SecureRandom.hex(12) item["queue"] ||= "default" Faktory.client_middleware.invoke(item, pool) do pool.with do |c| c.push(item) end end end |
#drain ⇒ Object
Drain and run all jobs for this worker
279 280 281 282 283 284 285 |
# File 'lib/faktory/testing.rb', line 279 def drain while jobs.any? next_job = jobs.first Queues.delete_for(next_job["jid"], next_job["queue"], self.to_s) process_job(next_job) end end |
#execute_job(worker, args) ⇒ Object
304 305 306 |
# File 'lib/faktory/testing.rb', line 304 def execute_job(worker, args) worker.perform(*args) end |
#faktory_class_attribute(*attrs) ⇒ Object
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 |
# File 'lib/faktory/job.rb', line 122 def faktory_class_attribute(*attrs) instance_reader = true instance_writer = true attrs.each do |name| singleton_class.instance_eval do undef_method(name) if method_defined?(name) || private_method_defined?(name) end define_singleton_method(name) { nil } ivar = "@#{name}" singleton_class.instance_eval do m = "#{name}=" undef_method(m) if method_defined?(m) || private_method_defined?(m) end define_singleton_method("#{name}=") do |val| singleton_class.class_eval do undef_method(name) if method_defined?(name) || private_method_defined?(name) define_method(name) { val } end if singleton_class? class_eval do undef_method(name) if method_defined?(name) || private_method_defined?(name) define_method(name) do if instance_variable_defined? ivar instance_variable_get ivar else singleton_class.send name end end end end val end if instance_reader undef_method(name) if method_defined?(name) || private_method_defined?(name) define_method(name) do if instance_variable_defined?(ivar) instance_variable_get ivar else self.class.public_send name end end end if instance_writer m = "#{name}=" undef_method(m) if method_defined?(m) || private_method_defined?(m) attr_writer name end end end |
#faktory_options(opts = {}) ⇒ Object
Allows customization of Faktory features for this type of Job. Legal options:
queue - use a named queue for this Job, default 'default'
retry - enable automatic retry for this Job, *Integer* count, default 25
backtrace - whether to save the error backtrace in the job payload to display in web UI,
an integer number of lines to save, default *0*
96 97 98 99 |
# File 'lib/faktory/job.rb', line 96 def (opts={}) # stringify self. = .merge(Hash[opts.map{|k, v| [k.to_s, v]}]) end |
#get_faktory_options ⇒ Object
:nodoc:
101 102 103 |
# File 'lib/faktory/job.rb', line 101 def # :nodoc: self. ||= Faktory. end |
#jobs ⇒ Object
Jobs queued for this worker
269 270 271 |
# File 'lib/faktory/testing.rb', line 269 def jobs Queues.jobs_by_worker[self.to_s] end |
#perform_async(*args) ⇒ Object
70 71 72 |
# File 'lib/faktory/job.rb', line 70 def perform_async(*args) client_push('jobtype'.freeze => self, 'args'.freeze => args) end |
#perform_in(interval, *args) ⇒ Object Also known as: perform_at
interval
must be a timestamp, numeric or something that acts
numeric (like an activesupport time interval).
76 77 78 79 80 81 82 83 84 |
# File 'lib/faktory/job.rb', line 76 def perform_in(interval, *args) int = interval.to_f now = Time.now.to_f ts = (int < 1_000_000_000 ? now + int : int) item = { 'jobtype'.freeze => self, 'args'.freeze => args } item['at'] = Time.at(ts).utc.to_datetime.rfc3339(9) if ts > now client_push(item) end |
#perform_one ⇒ Object
Pop out a single job and perform it
288 289 290 291 292 293 |
# File 'lib/faktory/testing.rb', line 288 def perform_one raise(EmptyQueueError, "perform_one called with empty job queue") if jobs.empty? next_job = jobs.first Queues.delete_for(next_job["jid"], queue, self.to_s) process_job(next_job) end |
#process_job(job) ⇒ Object
295 296 297 298 299 300 301 302 |
# File 'lib/faktory/testing.rb', line 295 def process_job(job) worker = new worker.jid = job['jid'] worker.bid = job['bid'] if worker.respond_to?(:bid=) #Faktory::Testing.server_middleware.invoke(worker, job, job['queue']) do execute_job(worker, job['args']) #end end |
#queue ⇒ Object
Queue for this worker
264 265 266 |
# File 'lib/faktory/testing.rb', line 264 def queue self.["queue"] end |