Class: Kaal::Registry

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/kaal/registry.rb

Overview

Thread-safe registry for storing and managing registered cron jobs. Each entry consists of a unique key, cron expression, and enqueue callback.

Examples:

Register a job

registry = Kaal::Registry.new
registry.add(key: "reports:daily", cron: "0 9 * * *", enqueue: ->(fire_time:, idempotency_key:) { })

Retrieve all jobs

registry.all # => [#<Kaal::Registry::Entry key="reports:daily", cron="0 9 * * *", enqueue=#<Proc ...>>]

Defined Under Namespace

Classes: Entry

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Registry

Initialize a new Registry instance.



27
28
29
30
 
# File 'lib/kaal/registry.rb', line 27

def initialize
  @entries = {}
  @mutex = Mutex.new
end

Instance Method Details

#add(key:, cron:, enqueue:) ⇒ Entry

Register a new cron job.

Entries are frozen after creation to prevent external mutation that could corrupt the internal key->entry mapping.

Examples:

registry.add(
  key: "job:daily",
  cron: "0 9 * * *",
  enqueue: ->(fire_time:, idempotency_key:) { MyJob.perform_later }
)

Parameters:

  • key (String) —

    unique identifier for the cron task

  • cron (String) —

    cron expression (e.g., “0 9 * * *”, “@daily”)

  • enqueue (Proc, Lambda) —

    callable that executes when cron fires

Returns:

  • (Entry) —

    the registered entry (frozen)

Raises:

  • (ArgumentError) —

    if key is empty, cron is empty, or enqueue is not callable

  • (RegistryError) —

    if key is already registered



52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/kaal/registry.rb', line 52

def add(key:, cron:, enqueue:)
  validate_entry(key, cron, enqueue)

  @mutex.synchronize do
    raise RegistryError, "Key '#{key}' is already registered" if @entries.key?(key)

    entry = Entry.new(key: key, cron: cron, enqueue: enqueue).freeze
    @entries[key] = entry
    entry
  end
end

#all ⇒ Array<Entry>

Get all registered entries.

Examples:

all_entries = registry.all
all_entries.each { |entry| puts entry.key }

Returns:

  • (Array<Entry>) —

    a copy of all registered entries



121
122
123
124
125
 
# File 'lib/kaal/registry.rb', line 121

def all
  @mutex.synchronize do
    @entries.values.dup
  end
end

#clear ⇒ Integer

Clear all registered entries.

Examples:

cleared_count = registry.clear

Returns:

  • (Integer) —

    the number of entries that were cleared



167
168
169
170
171
172
173
 
# File 'lib/kaal/registry.rb', line 167

def clear
  @mutex.synchronize do
    count = @entries.size
    @entries.clear
    count
  end
end

#each {|entry| ... } ⇒ void

This method returns an undefined value.

Iterate over all entries with thread-safe access.

Copies entries inside the lock and yields outside to avoid deadlocks if the block calls back into the registry.

Examples:

registry.each { |entry| puts entry.key }

Yields:

  • (entry) —

    yields each entry to the block

Yield Parameters:

  • entry (Entry) —

    the cron entry



187
188
189
190
191
192
193
194
195
 
# File 'lib/kaal/registry.rb', line 187

def each(&)
  return enum_for(:each) unless block_given?

  entries_snapshot = @mutex.synchronize do
    @entries.values.dup
  end

  entries_snapshot.each(&)
end

#find(key) ⇒ Entry?

Find a registered entry by key.

Examples:

entry = registry.find("job:daily")
entry.cron # => "0 9 * * *"

Parameters:

  • key (String) —

    the key to look up

Returns:

  • (Entry, nil) —

    the entry if found, nil otherwise



107
108
109
110
111
 
# File 'lib/kaal/registry.rb', line 107

def find(key)
  @mutex.synchronize do
    @entries[key]
  end
end

#inspect ⇒ String

Get a string representation of the registry.

Returns:

  • (String) —

    human-readable registry summary



214
215
216
217
218
 
# File 'lib/kaal/registry.rb', line 214

def inspect
  @mutex.synchronize do
    "#<Kaal::Registry size=#{@entries.size} keys=[#{@entries.keys.map(&:inspect).join(', ')}]>"
  end
end

#registered?(key) ⇒ Boolean

Check if a key is registered.

Examples:

registry.registered?("job:daily") # => true

Parameters:

  • key (String) —

    the key to check

Returns:

  • (Boolean) —

    true if the key is registered, false otherwise



154
155
156
157
158
 
# File 'lib/kaal/registry.rb', line 154

def registered?(key)
  @mutex.synchronize do
    @entries.key?(key)
  end
end

#remove(key) ⇒ Entry?

Unregister (remove) a cron job by key.

Examples:

registry.remove("job:daily")

Parameters:

  • key (String) —

    the key to unregister

Returns:

  • (Entry, nil) —

    the removed entry, or nil if not found



92
93
94
95
96
 
# File 'lib/kaal/registry.rb', line 92

def remove(key)
  @mutex.synchronize do
    @entries.delete(key)
  end
end

#size ⇒ Integer Also known as: count

Get the number of registered entries.

Examples:

registry.size # => 3

Returns:

  • (Integer) —

    the count of registered cron jobs



134
135
136
137
138
 
# File 'lib/kaal/registry.rb', line 134

def size
  @mutex.synchronize do
    @entries.size
  end
end

#to_a ⇒ Array<Hash>

Convert registry to an array of hashes.

Examples:

registry.to_a # => [{ key: "job:daily", cron: "0 9 * * *", enqueue: Proc }]

Returns:

  • (Array<Hash>) —

    array of entry details



204
205
206
207
208
 
# File 'lib/kaal/registry.rb', line 204

def to_a
  @mutex.synchronize do
    @entries.values.map(&:to_h)
  end
end

#upsert(key:, cron:, enqueue:) ⇒ Entry

Insert or replace a cron job entry atomically.

Unlike #add, this does not raise when the key already exists.

Parameters:

  • key (String) —

    unique identifier for the cron task

  • cron (String) —

    cron expression (e.g., “0 9 * * *”, “@daily”)

  • enqueue (Proc, Lambda) —

    callable that executes when cron fires

Returns:

  • (Entry) —

    the stored entry (frozen)

Raises:

  • (ArgumentError) —

    if key is empty, cron is empty, or enqueue is not callable



75
76
77
78
79
80
81
82
 
# File 'lib/kaal/registry.rb', line 75

def upsert(key:, cron:, enqueue:)
  validate_entry(key, cron, enqueue)

  @mutex.synchronize do
    entry = Entry.new(key: key, cron: cron, enqueue: enqueue).freeze
    @entries[key] = entry
  end
end