Class: Backburner::Workers::Forking

Inherits:
Backburner::Worker show all
Defined in:
lib/backburner/workers/forking.rb

Instance Attribute Summary

Attributes inherited from Backburner::Worker

#connection, #tube_names

Instance Method Summary collapse

Methods inherited from Backburner::Worker

enqueue, #handle_failure_for_job, #initialize, #process_tube_names, #shutdown, start, #work_one_job

Methods included from Logger

included, #job_started_at, #log_error, #log_info, #log_job_begin, #log_job_end, #logger

Methods included from Helpers

#classify, #constantize, #dasherize, #exception_message, #expand_tube_name, included, #queue_config, #resolve_max_job_retries, #resolve_priority, #resolve_respond_timeout, #resolve_retry_delay, #resolve_retry_delay_proc

Constructor Details

This class inherits a constructor from Backburner::Worker

Instance Method Details

#coolest_exit ⇒ Object

Exit with Kernel.exit! to avoid at_exit callbacks that should belongs to parent process We will use exitcode 99 that means the fork reached the garbage number



46
47
48
 
# File 'lib/backburner/workers/forking.rb', line 46

def coolest_exit
  Kernel.exit! 99
end

#fork_one_job ⇒ Object

Need to re-establish the connection to the server(s) after forking Waits for a job, works the job, and exits



30
31
32
33
34
35
36
 
# File 'lib/backburner/workers/forking.rb', line 30

def fork_one_job
  pid = Process.fork do
    work_one_job
    coolest_exit
  end
  Process.wait(pid)
end

#on_reconnect(conn) ⇒ Object 



38
39
40
41
 
# File 'lib/backburner/workers/forking.rb', line 38

def on_reconnect(conn)
  @connection = conn
  prepare
end

#prepare ⇒ Object

Used to prepare job queues before processing jobs. Setup beanstalk tube_names and watch all specified tubes for jobs.

Examples:

@worker.prepare

Raises:

  • (Beaneater::NotConnected) —

    If beanstalk fails to connect.



11
12
13
14
15
 
# File 'lib/backburner/workers/forking.rb', line 11

def prepare
  self.tube_names.map! { |name| expand_tube_name(name)  }.uniq!
  log_info "Working #{tube_names.size} queues: [ #{tube_names.join(', ')} ]"
#        self.connection.tubes.watch!(*self.tube_names)
end

#start ⇒ Object

Starts processing new jobs indefinitely. Primary way to consume and process jobs in specified tubes.

Examples:

@worker.start


23
24
25
26
 
# File 'lib/backburner/workers/forking.rb', line 23

def start
  prepare
  loop { fork_one_job }
end