class Rufus::Scheduler::Job

Attributes

callable[R]

anything with a call(job[, timet]) method, what gets actually triggered

count[R]
handler[R]

a reference to the instance whose call method is the @callable

id[R]
job_id[R]
last_time[R]
last_work_time[R]
mean_work_time[R]
next_time[RW]

next trigger time

opts[R]
original[R]
previous_time[RW]

previous “next trigger time”

scheduled_at[R]
tags[R]
unscheduled_at[R]

Public Class Methods

new(scheduler, original, opts, block) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 45
def initialize(scheduler, original, opts, block)

  @scheduler = scheduler
  @original = original
  @opts = opts

  @handler = block

  @callable =
    if block.respond_to?(:arity)
      block
    elsif block.respond_to?(:call)
      block.method(:call)
    elsif block.is_a?(Class)
      @handler = block.new
      @handler.method(:call) rescue nil
    else
      nil
    end

  @scheduled_at = EoTime.now
  @unscheduled_at = nil
  @last_time = nil

  @locals = {}
  @local_mutex = Mutex.new

  @id = determine_id

  fail(
    ArgumentError,
    'missing block or callable to schedule',
    caller[2..-1]
  ) unless @callable

  @tags = Array(opts[:tag] || opts[:tags]).collect { |t| t.to_s }

  @count = 0
  @last_work_time = 0.0
  @mean_work_time = 0.0

  # tidy up options

  if @opts[:allow_overlap] == false || @opts[:allow_overlapping] == false
    @opts[:overlap] = false
  end
  if m = @opts[:mutex]
    @opts[:mutex] = Array(m)
  end
end

Public Instance Methods

[](key) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 148
def [](key)

  @local_mutex.synchronize { @locals[key] }
end
[]=(key, value) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 143
def []=(key, value)

  @local_mutex.synchronize { @locals[key] = value }
end
call(do_rescue=false) click to toggle source

Calls the callable (usually a block) wrapped in this Job instance.

Warning: error rescueing is the responsibity of the caller.

# File lib/rufus/scheduler/jobs.rb, line 176
def call(do_rescue=false)

  do_call(EoTime.now, do_rescue)
end
key?(key) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 153
def key?(key)

  @local_mutex.synchronize { @locals.key?(key) }
end
keys() click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 158
def keys

  @local_mutex.synchronize { @locals.keys }
end
kill() click to toggle source

Kills all the threads this Job currently has going on.

# File lib/rufus/scheduler/jobs.rb, line 128
def kill

  threads.each { |t| t.raise(KillSignal) }
end
running?() click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 133
def running?

  threads.any?
end
scheduled?() click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 138
def scheduled?

  @scheduler.scheduled?(self)
end
threads() click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 121
def threads

  Thread.list.select { |t| t[:rufus_scheduler_job] == self }
end
trigger(time) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 98
def trigger(time)

  @previous_time = @next_time
  set_next_time(time)

  do_trigger(time)
end
trigger_off_schedule(time=EoTime.now) click to toggle source

Trigger the job right now, off of its schedule.

Done in collaboration with Piavka in github.com/jmettraux/rufus-scheduler/issues/214

# File lib/rufus/scheduler/jobs.rb, line 111
def trigger_off_schedule(time=EoTime.now)

  do_trigger(time)
end
unschedule() click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 116
def unschedule

  @unscheduled_at = EoTime.now
end

Protected Instance Methods

callback(meth, time) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 183
def callback(meth, time)

  return true unless @scheduler.respond_to?(meth)

  arity = @scheduler.method(meth).arity
  args = [ self, time ][0, (arity < 0 ? 2 : arity)]

  @scheduler.send(meth, *args)
end
compute_timeout() click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 193
def compute_timeout

  if to = @opts[:timeout]
    Rufus::Scheduler.parse(to)
  else
    nil
  end
end
do_call(time, do_rescue) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 207
def do_call(time, do_rescue)

  args = [ self, time ][0, @callable.arity]
  @callable.call(*args)

rescue StandardError => se

  fail se unless do_rescue

  return if se.is_a?(KillSignal) # discard

  @scheduler.on_error(self, se)

# exceptions above StandardError do pass through
end
do_trigger(time) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 223
def do_trigger(time)

  return if (
    opts[:overlap] == false &&
    running?
  )
  return if (
    callback(:confirm_lock, time) &&
    callback(:on_pre_trigger, time)
  ) == false

  @count += 1

  if opts[:blocking]
    trigger_now(time)
  else
    trigger_queue(time)
  end
end
mutex(m) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 202
def mutex(m)

  m.is_a?(Mutex) ? m : (@scheduler.mutexes[m.to_s] ||= Mutex.new)
end
post_trigger(time) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 270
def post_trigger(time)

  set_next_time(time, true)

  callback(:on_post_trigger, time)
end
start_work_thread() click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 277
def start_work_thread

  thread =
    Thread.new do

      Thread.current[@scheduler.thread_key] = true
      Thread.current[:rufus_scheduler_work_thread] = true

      loop do

        job, time = @scheduler.work_queue.pop

        break if @scheduler.started_at == nil

        next if job.unscheduled_at

        begin

          (job.opts[:mutex] || []).reduce(
            lambda { job.trigger_now(time) }
          ) do |b, m|
            lambda { mutex(m).synchronize { b.call } }
          end.call

        rescue KillSignal

          # simply go on looping
        end
      end
    end

  thread[@scheduler.thread_key] = true
  thread[:rufus_scheduler_work_thread] = true
    #
    # same as above (in the thead block),
    # but since it has to be done as quickly as possible.
    # So, whoever is running first (scheduler thread vs job thread)
    # sets this information

  thread
end
trigger_now(time) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 243
def trigger_now(time)

  t = EoTime.now
    # if there are mutexes, t might be really bigger than time

  Thread.current[:rufus_scheduler_job] = self
  Thread.current[:rufus_scheduler_time] = t
  Thread.current[:rufus_scheduler_timeout] = compute_timeout

  @last_time = t

  do_call(time, true)

ensure

  @last_work_time =
    EoTime.now - Thread.current[:rufus_scheduler_time]
  @mean_work_time =
    ((@count - 1) * @mean_work_time + @last_work_time) / @count

  post_trigger(time)

  Thread.current[:rufus_scheduler_job] = nil
  Thread.current[:rufus_scheduler_time] = nil
  Thread.current[:rufus_scheduler_timeout] = nil
end
trigger_queue(time) click to toggle source
# File lib/rufus/scheduler/jobs.rb, line 319
def trigger_queue(time)

  threads = @scheduler.work_threads

  cur = threads.size
  vac = threads.select { |t| t[:rufus_scheduler_job] == nil }.size
  #min = @scheduler.min_work_threads
  max = @scheduler.max_work_threads
  que = @scheduler.work_queue.size

  start_work_thread if vac - que < 1 && cur < max

  @scheduler.work_queue << [ self, time ]
end