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, &block) ⇒ Object 



31
32
33
34
35
36
37
38
39
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 31

def every(expression, name: nil, timeout: nil, overlap: true, &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) 


51
52
53
54
55
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 51

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

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

#job_names ⇒ Object 



41
42
43
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 41

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

#next_runs(from: Time.now) ⇒ Object 



83
84
85
86
87
88
89
90
91
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 83

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(&block) ⇒ Object 



21
22
23
24
25
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 21

def on_error(&block)
  return @on_error unless block

  @on_error = block
end

#remove(name) ⇒ Object 



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

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



75
76
77
78
79
80
81
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 75

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

Returns:

  • (Boolean) 


114
115
116
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 114

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

#running_jobs ⇒ Object 



27
28
29
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 27

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

#start ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 93

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 ⇒ Object 



106
107
108
109
110
111
112
 
# File 'lib/philiprehberger/cron_kit/scheduler.rb', line 106

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

  self
end