Class: Philiprehberger::CronKit::Scheduler

Inherits:
Object
  • Object
show all
Includes:
TimeoutHandler
Defined in:
lib/philiprehberger/cron_kit/scheduler.rb

Overview

A simple in-process cron scheduler that checks registered jobs every 60 seconds.

Defined Under Namespace

Classes: Job

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Scheduler

Returns a new instance of Scheduler.



11
12
13
14
15
16
17
18
19
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 11

def initialize
  @jobs = []
  @mutex = Mutex.new
  @thread = nil
  @running = false
  @on_error = nil
  @running_threads = []
  @job_threads = {}
end

Instance Method Details

#every(expression, name: nil, timeout: nil, overlap: true) {|time| ... } ⇒ self

Register a cron job to be executed when expression matches.

Accepts either a cron expression string or a pre-built Expression. Raises ArgumentError immediately if no block is provided, surfacing configuration mistakes at registration time rather than at tick time.

Parameters:

  • expression (String, Expression) —

    cron expression or pre-built Expression

  • name (Symbol, String, nil) (defaults to: nil) —

    optional job name for lookup/removal/triggering

  • timeout (Numeric, nil) (defaults to: nil) —

    kill the job if it exceeds this many seconds

  • overlap (Boolean) (defaults to: true) —

    when false, skip the tick if a previous run is still active

Yield Parameters:

  • time (Time) —

    the tick time the job fires at

Returns:

  • (self)

Raises:

  • (ArgumentError) —

    if no block is provided



55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 55

def every(expression, name: nil, timeout: nil, overlap: true, &block)
  raise ArgumentError, 'block required' unless block

  expr = expression.is_a?(Expression) ? expression : Expression.new(expression)

  @mutex.synchronize do
    @jobs << Job.new(expression: expr, block: block, name: name, timeout: timeout, overlap: overlap)
  end

  self
end

#job?(name) ⇒ Boolean

Whether a job is registered under name.

Anonymous jobs (registered without a name:) are never matched.

Parameters:

  • name (Symbol, String, nil)

Returns:

  • (Boolean) 


82
83
84
85
86
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 82

def job?(name)
  return false if name.nil?

  @mutex.synchronize { @jobs.any? { |j| j.name == name } }
end

#job_names ⇒ Array<Symbol, String>

Names of all registered named jobs.

Anonymous jobs (registered without name:) are omitted.

Returns:

  • (Array<Symbol, String>) —

    list of registered job names



72
73
74
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 72

def job_names
  @mutex.synchronize { @jobs.map(&:name).compact }
end

#next_runs(from: Time.now) ⇒ Hash{Symbol, String => Time}

Upcoming run times for every named job.

Anonymous jobs are omitted.

Parameters:

  • from (Time) (defaults to: Time.now) —

    starting point for the lookahead (defaults to Time.now)

Returns:

  • (Hash{Symbol, String => Time}) —

    map of job name to the next matching time



124
125
126
127
128
129
130
131
132
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 124

def next_runs(from: Time.now)
  jobs = @mutex.synchronize { @jobs.dup }

  jobs.each_with_object({}) do |job, hash|
    next unless job.name

    hash[job.name] = job.expression.next_at(from: from)
  end
end

#on_error {|job, error| ... } ⇒ Proc?

Register a callback invoked when a job raises an error.

Called with the failing Job and the StandardError raised. With no block, returns the currently registered callback (or nil).

Yield Parameters:

  • job (Job) —

    the job that failed

  • error (StandardError) —

    the error raised by the job’s block

Returns:

  • (Proc, nil) —

    the registered callback when called without a block



29
30
31
32
33
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 29

def on_error(&block)
  return @on_error unless block

  @on_error = block
end

#remove(name) ⇒ Boolean

Remove a registered job by name.

Parameters:

  • name (Symbol, String) —

    the job name to remove

Returns:

  • (Boolean) —

    true if a matching job was removed, false otherwise



92
93
94
95
96
97
98
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 92

def remove(name)
  @mutex.synchronize do
    initial_size = @jobs.size
    @jobs.reject! { |job| job.name == name }
    @jobs.size < initial_size
  end
end

#run_now(name) ⇒ Object?

Manually trigger a registered job by name.

Reuses the standard execution path (timeout + overlap-skip), but runs synchronously so the caller can capture the return value. Useful for testing and operator-driven re-runs.

Parameters:

  • name (Symbol, String) —

    the job name to trigger

Returns:

  • (Object, nil) —

    the block’s return value, or ‘nil` when skipped due to `overlap: false`

Raises:

  • (KeyError) —

    if no job is registered under ‘name`

  • (Timeout::Error) —

    if the job’s ‘timeout:` is exceeded



110
111
112
113
114
115
116
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 110

def run_now(name)
  job = @mutex.synchronize { @jobs.find { |j| j.name == name } }
  raise KeyError, "job not registered: #{name.inspect}" unless job
  return nil if skip_overlapping?(job)

  execute_now(job)
end

#running? ⇒ Boolean

Whether the scheduler loop is currently running.

Returns:

  • (Boolean) 


166
167
168
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 166

def running?
  @mutex.synchronize { @running }
end

#running_jobs ⇒ Integer

Number of currently executing job threads.

Returns:

  • (Integer) —

    count of live job threads



38
39
40
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 38

def running_jobs
  @mutex.synchronize { @running_threads.count(&:alive?) }
end

#start ⇒ self

Start the scheduler loop in a background thread.

Idempotent — calling start on an already-running scheduler is a no-op.

Returns:

  • (self) 


139
140
141
142
143
144
145
146
147
148
149
150
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 139

def start
  @mutex.synchronize do
    return self if @running

    @running = true
  end

  @thread = Thread.new { run_loop }
  @thread.abort_on_exception = true

  self
end

#stop ⇒ self

Stop the scheduler loop and wait briefly for the background thread to exit.

Returns:

  • (self) 


155
156
157
158
159
160
161
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 155

def stop
  @mutex.synchronize { @running = false }
  @thread&.join(5)
  @thread = nil

  self
end