Module: RocketJob

Defined in:
lib/rocketjob.rb,
lib/rocket_job/cli.rb,
lib/rocket_job/job.rb,
lib/rocket_job/batch.rb,
lib/rocket_job/event.rb,
lib/rocket_job/config.rb,
lib/rocket_job/server.rb,
lib/rocket_job/sliced.rb,
lib/rocket_job/worker.rb,
lib/rocket_job/railtie.rb,
lib/rocket_job/version.rb,
lib/rocket_job/batch/io.rb,
lib/rocket_job/heartbeat.rb,
lib/rocket_job/rocket_job.rb,
lib/rocket_job/subscriber.rb,
lib/rocket_job/supervisor.rb,
lib/rocket_job/batch/model.rb,
lib/rocket_job/performance.rb,
lib/rocket_job/worker_pool.rb,
lib/rocket_job/batch/logger.rb,
lib/rocket_job/batch/result.rb,
lib/rocket_job/batch/worker.rb,
lib/rocket_job/dirmon_entry.rb,
lib/rocket_job/plugins/cron.rb,
lib/rocket_job/server/model.rb,
lib/rocket_job/sliced/input.rb,
lib/rocket_job/sliced/slice.rb,
lib/rocket_job/active_worker.rb,
lib/rocket_job/batch/results.rb,
lib/rocket_job/category/base.rb,
lib/rocket_job/job_exception.rb,
lib/rocket_job/plugins/retry.rb,
lib/rocket_job/ractor_worker.rb,
lib/rocket_job/sliced/output.rb,
lib/rocket_job/sliced/slices.rb,
lib/rocket_job/thread_worker.rb,
lib/rocket_job/batch/throttle.rb,
lib/rocket_job/category/input.rb,
lib/rocket_job/batch/callbacks.rb,
lib/rocket_job/category/output.rb,
lib/rocket_job/jobs/active_job.rb,
lib/rocket_job/jobs/dirmon_job.rb,
lib/rocket_job/jobs/simple_job.rb,
lib/rocket_job/plugins/restart.rb,
lib/rocket_job/batch/categories.rb,
lib/rocket_job/batch/statistics.rb,
lib/rocket_job/plugins/document.rb,
lib/rocket_job/batch/performance.rb,
lib/rocket_job/lookup_collection.rb,
lib/rocket_job/plugins/job/model.rb,
lib/rocket_job/plugins/singleton.rb,
lib/rocket_job/jobs/copy_file_job.rb,
lib/rocket_job/jobs/on_demand_job.rb,
lib/rocket_job/plugins/job/logger.rb,
lib/rocket_job/plugins/job/worker.rb,
lib/rocket_job/subscribers/logger.rb,
lib/rocket_job/subscribers/server.rb,
lib/rocket_job/subscribers/worker.rb,
lib/rocket_job/batch/state_machine.rb,
lib/rocket_job/jobs/conversion_job.rb,
lib/rocket_job/plugins/transaction.rb,
lib/rocket_job/sliced/writer/input.rb,
lib/rocket_job/supervisor/shutdown.rb,
lib/rocket_job/throttle_definition.rb,
lib/rocket_job/batch/lower_priority.rb,
lib/rocket_job/jobs/performance_job.rb,
lib/rocket_job/jobs/upload_file_job.rb,
lib/rocket_job/plugins/job/throttle.rb,
lib/rocket_job/server/state_machine.rb,
lib/rocket_job/sliced/writer/output.rb,
lib/rocket_job/throttle_definitions.rb,
lib/rocket_job/jobs/housekeeping_job.rb,
lib/rocket_job/plugins/job/callbacks.rb,
lib/rocket_job/plugins/state_machine.rb,
lib/rocket_job/batch/throttle_windows.rb,
lib/rocket_job/sliced/encrypted_slice.rb,
lib/rocket_job/plugins/job/persistence.rb,
lib/rocket_job/sliced/compressed_slice.rb,
lib/rocket_job/jobs/on_demand_batch_job.rb,
lib/rocket_job/plugins/job/state_machine.rb,
lib/rocket_job/plugins/processing_window.rb,
lib/rocket_job/sliced/bzip2_output_slice.rb,
lib/rocket_job/subscribers/secret_config.rb,
lib/rocket_job/extensions/mongoid/factory.rb,
lib/rocket_job/batch/throttle_running_workers.rb,
lib/rocket_job/jobs/re_encrypt/relational_job.rb,
lib/rocket_job/plugins/throttle_dependent_jobs.rb,
lib/rocket_job/plugins/job/throttle_running_jobs.rb,
lib/rocket_job/extensions/mongoid/clients/options.rb,
lib/rocket_job/sliced/encrypted_bzip2_output_slice.rb

Overview

Job to dynamically perform ruby code on demand as a Batch,

Often used for data correction or cleansing.

Example: Iterate over all rows in a table:

code = <<-CODE
  if user = User.find(row)
    user.cleanse_attributes!
    user.save(validate: false)
  end
CODE
job  = RocketJob::Jobs::OnDemandBatchJob.new(code: code, description: 'cleanse users')
arel = User.unscoped.all.order('updated_at DESC')
job.upload_arel(arel)
job.save!

Console Testing:

code = <<-CODE
  if user = User.find(row)
    user.cleanse_attributes!
    user.save(validate: false)
  end
CODE
job  = RocketJob::Jobs::OnDemandBatchJob.new(code: code, description: 'cleanse users')

# Run against a sub-set using a limit
arel = User.unscoped.all.order('updated_at DESC').limit(100)
job.upload_arel(arel)

# Run the subset directly within the console
job.perform_now
job.cleanup!

By default output is not collected, call the method ‘#collect_output` to collect output.

Example:

job = RocketJob::Jobs::OnDemandBatchJob(
  description:              'Fix data',
  code:                     code,
  throttle_running_workers: 5,
  priority:                 30
)
job.collect_output
job.save!

Example: Move the upload operation into a before_batch.

upload_code = <<-CODE
  arel = User.unscoped.all.order('updated_at DESC')
  upload_arel(arel)
CODE

code = <<-CODE
  if user = User.find(row)
    user.cleanse_attributes!
    user.save(validate: false)
  end
CODE

RocketJob::Jobs::OnDemandBatchJob.create!(
  upload_code: upload_code,
  code:        code,
  description: 'cleanse users'
)

Defined Under Namespace

Modules: Batch, Category, Jobs, MongoidClients, MongoidFactory, Plugins, Sliced, Subscriber, Subscribers Classes: ActiveWorker, CLI, Config, DirmonEntry, Event, Heartbeat, Job, JobException, LookupCollection, Performance, RactorWorker, Railtie, Server, Supervisor, ThreadWorker, ThrottleDefinition, ThrottleDefinitions, Worker, WorkerPool

Constant Summary collapse

VERSION = 
"6.3.1".freeze

Class Method Summary collapse

Class Method Details

.create_indexes ⇒ Object 



2
3
4
5
6
7
 
# File 'lib/rocket_job/rocket_job.rb', line 2

def self.create_indexes
  # Ensure models with indexes are loaded into memory first
  Job.create_indexes
  Server.create_indexes
  DirmonEntry.create_indexes
end

.rails! ⇒ Object 



57
58
59
 
# File 'lib/rocket_job/rocket_job.rb', line 57

def self.rails!
  @rails = true
end

.rails? ⇒ Boolean

When running inside a Rocket Job server process, returns true when Rails has been initialized.

Returns:

  • (Boolean) 


16
17
18
 
# File 'lib/rocket_job/rocket_job.rb', line 16

def self.rails?
  @rails
end

.seconds_as_duration(seconds) ⇒ Object

Returns a human readable duration from the supplied [Float] number of seconds



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/rocket_job/rocket_job.rb', line 27

def self.seconds_as_duration(seconds)
  return nil unless seconds

  if seconds >= 86_400.0 # 1 day
    "#{(seconds / 86_400).to_i}d #{Time.at(seconds).utc.strftime('%-Hh %-Mm')}"
  elsif seconds >= 3600.0 # 1 hour
    Time.at(seconds).utc.strftime("%-Hh %-Mm")
  elsif seconds >= 60.0 # 1 minute
    Time.at(seconds).utc.strftime("%-Mm %-Ss")
  elsif seconds >= 1.0 # 1 second
    format("%.3fs", seconds)
  else
    duration = seconds * 1000
    if defined? JRuby
      "#{duration.to_i}ms"
    else
      duration < 10.0 ? format("%.3fms", duration) : format("%.1fms", duration)
    end
  end
end

.server! ⇒ Object 



53
54
55
 
# File 'lib/rocket_job/rocket_job.rb', line 53

def self.server!
  @server = true
end

.server? ⇒ Boolean

Whether the current process is running inside a Rocket Job server process.

Returns:

  • (Boolean) 


10
11
12
 
# File 'lib/rocket_job/rocket_job.rb', line 10

def self.server?
  @server
end

.standalone? ⇒ Boolean

When running inside a Rocket Job server process, returns true when running standalone.

Returns:

  • (Boolean) 


22
23
24
 
# File 'lib/rocket_job/rocket_job.rb', line 22

def self.standalone?
  !@rails
end