mahlon/symphony-metronome
1
0
Fork 0
Code Issues Pull requests Activity

Commit the pile of initial work. Largely taken from company internal

Browse source 
repos, allowed to be newly released.  Portions already in production,
coverage still needs to be boosted.  Enjoy.

FossilOrigin-Name: 0f17fa483f55467bdf9e8f99dace58e6a90f5a8a7e595bdd79dfda5c92d16b7f
This commit is contained in:
Mahlon E. Smith 2014-04-22 00:21:43 +00:00
commit b18647f6a5
23 changed files with 2639 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

79
lib/symphony/metronome.rb Normal file
View file

@ -0,0 +1,79 @@
#!/usr/bin/env ruby
# vim: set nosta noet ts=4 sw=4:
require 'pathname'
require 'symphony' unless defined?( Symphony )
module Symphony::Metronome
extend Loggability,
Configurability
# Library version constant
VERSION = '0.1.0'
# Version-control revision constant
REVISION = %q$Revision$
# The name of the environment variable to check for config file overrides
CONFIG_ENV = 'METRONOME_CONFIG'
# The path to the default config file
DEFAULT_CONFIG_FILE = 'etc/config.yml'
# The data directory that contains migration files.
#
DATADIR = if ENV['METRONOME_DATADIR']
Pathname.new( ENV['METRONOME_DATADIR'] )
elsif Gem.datadir( 'symphony-metronome' )
Pathname.new( Gem.datadir('symphony-metronome') )
else
Pathname.new( __FILE__ ).dirname.parent.parent + 'data/symphony-metronome'
end
# Loggability API -- use symphony's logger
log_to :symphony
# Configurability API
config_key :metronome
### Get the loaded config (a Configurability::Config object)
def self::config
Configurability.loaded_config
end
### Load the specified +config_file+, install the config in all objects with
### Configurability, and call any callbacks registered via #after_configure.
def self::load_config( config_file=nil, defaults=nil )
config_file ||= ENV[ CONFIG_ENV ] || DEFAULT_CONFIG_FILE
defaults ||= Configurability.gather_defaults
config = Configurability::Config.load( config_file, defaults )
config.install
end
# The generic parse exception class.
class TimeParseError < ArgumentError; end
require 'symphony/metronome/scheduler'
require 'symphony/metronome/intervalexpression'
require 'symphony/metronome/scheduledevent'
require 'symphony/tasks/scheduletask'
###############
module_function
###############
### Convenience method for running the scheduler.
###
def run( &block )
raise LocalJumpError, "No block provided." unless block_given?
return Symphony::Metronome::Scheduler.run( &block )
end
end # Symphony::Metronome

590
lib/symphony/metronome/intervalexpression.rl Normal file
View file

@ -0,0 +1,590 @@
# vim: set noet nosta sw=4 ts=4 ft=ragel :
%%{
#
# Generate the actual code like so:
# ragel -R -T1 -Ls inputfile.rl
#
machine interval_expression;
########################################################################
### A C T I O N S
########################################################################
action set_mark { mark = p }
action set_valid { event.instance_variable_set( :@valid, true ) }
action set_invalid { event.instance_variable_set( :@valid, false ) }
action recurring { event.instance_variable_set( :@recurring, true ) }
action start_time {
time = event.send( :extract, mark, p - mark )
event.send( :set_starting, time, :time )
}
action start_interval {
interval = event.send( :extract, mark, p - mark )
event.send( :set_starting, interval, :interval )
}
action execute_time {
time = event.send( :extract, mark, p - mark )
event.send( :set_interval, time, :time )
}
action execute_interval {
interval = event.send( :extract, mark, p - mark )
event.send( :set_interval, interval, :interval )
}
action execute_multiplier {
multiplier = event.send( :extract, mark, p - mark ).sub( / times/, '' )
event.instance_variable_set( :@multiplier, multiplier.to_i )
}
action ending_time {
time = event.send( :extract, mark, p - mark )
event.send( :set_ending, time, :time )
}
action ending_interval {
interval = event.send( :extract, mark, p - mark )
event.send( :set_ending, interval, :interval )
}
########################################################################
### P R E P O S I T I O N S
########################################################################
recur_preposition = ( 'every' | 'each' | 'per' | 'once' ' per'? ) @recurring;
time_preposition = 'at' | 'on';
interval_preposition = 'in';
########################################################################
### K E Y W O R D S
########################################################################
interval_times =
( 'milli'? 'second' | 'minute' | 'hour' | 'day' | 'week' | 'month' | 'year' ) 's'?;
start_identifiers = ( 'start' | 'begin' 'n'? ) 'ing'?;
exec_identifiers = ('run' | 'exec' 'ute'?);
ending_identifiers = ( ('for' | 'until' | 'during') | ('end'|'finish'|'stop'|'complet' 'e'?) 'ing'? );
########################################################################
### T I M E S P E C S
########################################################################
# 1st
# 202nd
# 2015th
# ...
#
ordinals = (
( (digit+ - '1')? '1' 'st' ) |
( digit+?
( '1' digit 'th' ) | # all '11s'
( '2' 'nd' ) |
( '3' 'rd' ) |
( [0456789] 'th' )
)
);
# 2014-05-01
# 2014-05-01 15:00
# 2014-05-01 15:00:30
#
fulldate = digit{4} '-' digit{2} '-' digit{2}
( space+ digit{2} ':' digit{2} ( ':' digit{2} )? )?;
# 10am
# 2:45pm
#
time = digit{1,2} ( ':' digit{2} )? ( 'am' | 'pm' );
# union of the above
date_or_time = fulldate | time;
# 20 seconds
# 5 hours
# 1 hour
# 2.5 hours
# an hour
# a minute
# other minute
#
interval = (
(( 'a' 'n'? | [1-9][0-9]* ( '.' [0-9]+ )? ) | 'other' | ordinals ) space+
)? interval_times;
########################################################################
### A C T I O N C H A I N S
########################################################################
start_time = date_or_time >set_mark %start_time;
start_interval = interval >set_mark %start_interval;
start_expression = ( (time_preposition space+)? start_time ) |
( (interval_preposition space+)? start_interval );
execute_time = date_or_time >set_mark %/execute_time;
execute_interval = interval >set_mark %execute_interval;
execute_multiplier = ( digit+ space+ 'times' )
>set_mark %execute_multiplier @recurring;
execute_expression = (
# regular dates and intervals
( time_preposition space+ execute_time ) |
( ( interval_preposition | recur_preposition ) space+ execute_interval )
) | (
# count + interval (10 times every minute)
execute_multiplier space+ ( recur_preposition space+ )? execute_interval
) |
# count for 'timeboxed' intervals
execute_multiplier;
ending_time = date_or_time >set_mark %ending_time;
ending_interval = interval >set_mark %ending_interval;
ending_expression = ( (time_preposition space+)? ending_time ) |
( (interval_preposition space+)? ending_interval );
########################################################################
### M A C H I N E S
########################################################################
Start = (
start: start_identifiers space+ -> StartTime,
StartTime: start_expression -> final
);
Interval = (
start:
Decorators: ( exec_identifiers space+ )? -> ExecuteTime,
ExecuteTime: execute_expression -> final
);
Ending = (
start: space+ ending_identifiers space+ -> EndingTime,
EndingTime: ending_expression -> final
);
main := (
( (Start space+)? Interval Ending? ) |
( Interval ( space+ Start )? Ending? ) |
( Interval Ending space+ Start )
) %set_valid @!set_invalid;
}%%
require 'symphony' unless defined?( Symphony )
require 'symphony/metronome'
require 'symphony/metronome/mixins'
using Symphony::Metronome::TimeRefinements
### Parse natural English expressions of times and intervals.
###
### in 30 minutes
### once an hour
### every 15 minutes for 2 days
### at 2014-05-01
### at 2014-04-01 14:00:25
### at 2pm
### starting at 2pm once a day
### start in 1 hour from now run every 5 seconds end at 11:15pm
### every other hour
### once a day ending in 1 week
### run once a minute for an hour starting in 6 days
### 10 times a minute for 2 days
### run 45 times every hour
### 30 times per day
### start at 2010-01-02 run 12 times and end on 2010-01-03
### starting in an hour from now run 6 times a minute for 2 hours
### beginning a day from now, run 30 times per minute and finish in 2 weeks
### execute 12 times during the next 2 minutes
###
class Symphony::Metronome::IntervalExpression
include Comparable,
Symphony::Metronome::TimeFunctions
extend Loggability
log_to :symphony
# Ragel accessors are injected as class methods/variables for some reason.
%% write data;
# Words/phrases in the expression that we'll strip/ignore before parsing.
COMMON_DECORATORS = [ 'and', 'then', /\s+from now/, 'the next' ];
########################################################################
### C L A S S M E T H O D S
########################################################################
### Parse a schedule expression +exp+.
###
### Parsing defaults to Time.now(), but if passed a +time+ object,
### all contexual times (2pm) are relative to it. If you know when
### an expression was generated, you can 'reconstitute' an interval
### object this way.
###
def self::parse( exp, time=nil )
# Normalize the expression before parsing
#
exp = exp.downcase.
gsub( /(?:[^[a-z][0-9][\.\-:]\s]+)/, '' ). # . : - a-z 0-9 only
gsub( Regexp.union(COMMON_DECORATORS), '' ). # remove common decorator words
gsub( /\s+/, ' ' ). # collapse whitespace
gsub( /([:\-])+/, '\1' ). # collapse multiple - or : chars
gsub( /\.+$/, '' ) # trailing periods
event = new( exp, time || Time.now )
data = event.instance_variable_get( :@data )
# Ragel interface variables
#
key = ''
mark = 0
%% write init;
eof = pe
%% write exec;
# Attach final time logic and sanity checks.
event.send( :finalize )
return event
end
########################################################################
### I N S T A N C E M E T H O D S
########################################################################
### Instantiate a new TimeExpression, provided an +expression+ string
### that describes when this event will take place in natural english,
### and a +base+ Time to perform calculations against.
###
private_class_method :new
def initialize( expression, base ) # :nodoc:
@exp = expression
@data = expression.to_s.unpack( 'c*' )
@base = base
@valid = false
@recurring = false
@starting = nil
@interval = nil
@multiplier = nil
@ending = nil
end
######
public
######
# Is the schedule expression parsable?
attr_reader :valid
# Does this event repeat?
attr_reader :recurring
# The valid start time for the schedule (for recurring events)
attr_reader :starting
# The valid end time for the schedule (for recurring events)
attr_reader :ending
# The interval to wait before the event should be acted on.
attr_reader :interval
# An optional interval multipler for expressing counts.
attr_reader :multiplier
### If this interval is on a stack somewhere and ready to
### fire, is it okay to do so based on the specified
### expression criteria?
###
### Returns +true+ if it should fire, +false+ if it should not
### but could at a later attempt, and +nil+ if the interval has
### expired.
###
def fire?
now = Time.now
# Interval has expired.
return nil if self.ending && now > self.ending
# Interval is not yet in its current time window.
return false if self.starting - now > 0
# Looking good.
return true
end
### Just return the original event expression.
###
def to_s
return @exp
end
### Inspection string.
###
def inspect
return ( "<%s:0x%08x valid:%s recur:%s expression:%p " +
"starting:%p interval:%p ending:%p>" ) % [
self.class.name,
self.object_id * 2,
self.valid,
self.recurring,
self.to_s,
self.starting,
self.interval,
self.ending
]
end
### Comparable interface, order by interval, 'soonest' first.
###
def <=>( other )
return self.interval <=> other.interval
end
#########
protected
#########
### Given a +start+ and +ending+ scanner position,
### return an ascii representation of the data slice.
###
def extract( start, ending )
slice = @data[ start, ending ]
return '' unless slice
return slice.pack( 'c*' )
end
### Parse and set the starting attribute, given a +time_arg+
### string and the +type+ of string (interval or exact time)
###
def set_starting( time_arg, type )
start = self.get_time( time_arg, type )
@starting = start
# If start time is expressed as a post-conditional (we've
# already got an end time) we need to recalculate the end
# as an offset from the start. The original parsed ending
# arguments should have already been cached when it was
# previously set.
#
if self.ending && self.recurring
self.set_ending( *@ending_args )
end
return @starting
end
### Parse and set the interval attribute, given a +time_arg+
### string and the +type+ of string (interval or exact time)
###
### Perform consistency and sanity checks before returning an
### integer representing the amount of time needed to sleep before
### firing the event.
###
def set_interval( time_arg, type )
interval = nil
if self.starting && type == :time
raise Symphony::Metronome::TimeParseError, "That doesn't make sense, just use 'at [datetime]' instead"
else
interval = self.get_time( time_arg, type )
interval = interval - @base
end
@interval = interval
return @interval
end
### Parse and set the ending attribute, given a +time_arg+
### string and the +type+ of string (interval or exact time)
###
### Perform consistency and sanity checks before returning a
### Time object.
###
def set_ending( time_arg, type )
ending = nil
# Ending dates only make sense for recurring events.
#
if self.recurring
@ending_args = [ time_arg, type ] # squirrel away for post-set starts
# Make the interval an offset of the start time, instead of now.
#
# This is the contextual difference between:
# every minute until 6 hours from now (ending based on NOW)
# and
# starting in a year run every minute for 1 month (ending based on start time)
#
if self.starting && type == :interval
diff = self.parse_interval( time_arg )
ending = self.starting + diff
# (offset from now)
#
else
ending = self.get_time( time_arg, type )
end
# Check the end time is after the start time.
#
if self.starting && ending < self.starting
raise Symphony::Metronome::TimeParseError, "recurring event ends before it begins"
end
else
self.log.debug "Ignoring ending date, event is not recurring."
end
@ending = ending
return @ending
end
### Perform finishing logic and final sanity checks before returning
### a parsed object.
###
def finalize
raise Symphony::Metronome::TimeParseError, "unable to parse expression" unless self.valid
# Ensure start time is populated.
#
unless self.starting
if self.recurring
@starting = @base
else
raise Symphony::Metronome::TimeParseError, "non-deterministic expression" if self.interval.nil?
@starting = @base + self.interval
end
end
# Alter the interval if a multiplier was specified.
#
if self.multiplier
if self.ending
# Regular 'count' style multipler with end date.
# (run 10 times a minute for 2 days)
# Just divide the current interval by the count.
#
if self.interval
@interval = self.interval.to_f / self.multiplier
# Timeboxed multiplier (start [date] run 10 times end [date])
# Evenly spread the interval out over the time window.
#
else
diff = self.ending - self.starting
@interval = diff.to_f / self.multiplier
end
# Regular 'count' style multipler (run 10 times a minute)
# Just divide the current interval by the count.
#
else
raise Symphony::Metronome::TimeParseError, "An end date or interval is required" unless self.interval
@interval = self.interval.to_f / self.multiplier
end
end
end
### Given a +time_arg+ string and a type (:interval or :time),
### dispatch to the appropriate parser.
###
def get_time( time_arg, type )
time = nil
if type == :interval
secs = self.parse_interval( time_arg )
time = @base + secs if secs
end
if type == :time
time = self.parse_time( time_arg )
end
raise Symphony::Metronome::TimeParseError, "unable to parse time" if time.nil?
return time
end
### Parse a +time_arg+ string (anything parsable buy Time.parse())
### into a Time object.
###
def parse_time( time_arg )
time = Time.parse( time_arg, @base ) rescue nil
# Generated date is in the past.
#
if time && @base > time
# Ensure future dates for ambiguous times (2pm)
time = time + 1.day if time_arg.length < 8
# Still in the past, abandon all hope.
raise Symphony::Metronome::TimeParseError, "attempt to schedule in the past" if @base > time
end
self.log.debug "Parsed %p (time) to: %p" % [ time_arg, time ]
return time
end
### Parse a +time_arg+ interval string ("30 seconds") into an
### Integer.
###
def parse_interval( interval_arg )
duration, span = interval_arg.split( /\s+/ )
# catch the 'a' or 'an' case (ex: "an hour")
duration = 1 if duration.index( 'a' ) == 0
# catch the 'other' case, ie: 'every other hour'
duration = 2 if duration == 'other'
# catch the singular case (ex: "hour")
unless span
span = duration
duration = 1
end
use_milliseconds = span.sub!( 'milli', '' )
interval = calculate_seconds( duration.to_f, span.to_sym )
# milliseconds
interval = duration.to_f / 1000 if use_milliseconds
self.log.debug "Parsed %p (interval) to: %p" % [ interval_arg, interval ]
return interval
end
end # class TimeExpression

130
lib/symphony/metronome/mixins.rb Normal file
View file

@ -0,0 +1,130 @@
# -*- ruby -*-
#encoding: utf-8
# vim: set nosta noet ts=4 sw=4:
require 'symphony' unless defined?( Symphony )
require 'symphony/metronome' unless defined?( Symphony::Metronome )
module Symphony::Metronome
# Functions for time calculations
module TimeFunctions
###############
module_function
###############
### Calculate the (approximate) number of seconds that are in +count+ of the
### given +unit+ of time.
###
def calculate_seconds( count, unit )
return case unit
when :seconds, :second
count
when :minutes, :minute
count * 60
when :hours, :hour
count * 3600
when :days, :day
count * 86400
when :weeks, :week
count * 604800
when :fortnights, :fortnight
count * 1209600
when :months, :month
count * 2592000
when :years, :year
count * 31557600
else
raise ArgumentError, "don't know how to calculate seconds in a %p" % [ unit ]
end
end
end # module TimeFunctions
# Refinements to Numeric to add time-related convenience methods
module TimeRefinements
refine Numeric do
### Number of seconds (returns receiver unmodified)
def seconds
return self
end
alias_method :second, :seconds
### Returns number of seconds in <receiver> minutes
def minutes
return TimeFunctions.calculate_seconds( self, :minutes )
end
alias_method :minute, :minutes
### Returns the number of seconds in <receiver> hours
def hours
return TimeFunctions.calculate_seconds( self, :hours )
end
alias_method :hour, :hours
### Returns the number of seconds in <receiver> days
def days
return TimeFunctions.calculate_seconds( self, :day )
end
alias_method :day, :days
### Return the number of seconds in <receiver> weeks
def weeks
return TimeFunctions.calculate_seconds( self, :weeks )
end
alias_method :week, :weeks
### Returns the number of seconds in <receiver> fortnights
def fortnights
return TimeFunctions.calculate_seconds( self, :fortnights )
end
alias_method :fortnight, :fortnights
### Returns the number of seconds in <receiver> months (approximate)
def months
return TimeFunctions.calculate_seconds( self, :months )
end
alias_method :month, :months
### Returns the number of seconds in <receiver> years (approximate)
def years
return TimeFunctions.calculate_seconds( self, :years )
end
alias_method :year, :years
### Returns the Time <receiver> number of seconds before the
### specified +time+. E.g., 2.hours.before( header.expiration )
def before( time )
return time - self
end
### Returns the Time <receiver> number of seconds ago. (e.g.,
### expiration > 2.hours.ago )
def ago
return self.before( ::Time.now )
end
### Returns the Time <receiver> number of seconds after the given +time+.
### E.g., 10.minutes.after( header.expiration )
def after( time )
return time + self
end
### Return a new Time <receiver> number of seconds from now.
def from_now
return self.after( ::Time.now )
end
end # refine Numeric
end # module TimeRefinements
end # module Symphony::Metronome

174
lib/symphony/metronome/scheduledevent.rb Normal file
View file

@ -0,0 +1,174 @@
#!/usr/bin/env ruby
# vim: set nosta noet ts=4 sw=4:
require 'set'
require 'sequel'
require 'sqlite3'
require 'yajl'
require 'symphony/metronome'
Sequel.extension :migration
### A class the represents the relationship between an interval and
### an event.
###
class Symphony::Metronome::ScheduledEvent
extend Loggability, Configurability
include Comparable
log_to :symphony
config_key :metronome
# Configure defaults.
#
CONFIG_DEFAULTS = {
db: 'sqlite:///tmp/metronome.db',
splay: 0
}
class << self
# A Sequel-style DB connection URI.
attr_reader :db
# Adjust recurring intervals by a random window.
attr_reader :splay
end
######################################################################
# C L A S S M E T H O D S
######################################################################
### Configurability API.
###
def self::configure( config=nil )
config = self.defaults.merge( config || {} )
@db = Sequel.connect( config.delete(:db) )
@splay = config.delete( :splay )
# Ensure the database is current.
#
migrations_dir = Symphony::Metronome::DATADIR + 'migrations'
unless Sequel::Migrator.is_current?( self.db, migrations_dir.to_s )
Sequel::Migrator.apply( self.db, migrations_dir.to_s )
end
end
### Return a set of all known events, sorted by date of execution.
### Delete any rows that are invalid expressions.
###
def self::load
now = Time.now
events = SortedSet.new
# Force reset the DB handle.
self.db.disconnect
self.log.debug "Parsing/loading all actions."
self.db[ :metronome ].each do |event|
begin
event = new( event )
events << event
rescue ArgumentError, Symphony::Metronome::TimeParseError => err
self.log.error "%p while parsing \"%s\": %s" % [
err.class,
event[:expression],
err.message
]
self.log.debug " " + err.backtrace.join( "\n " )
self.db[ :metronome ].filter( :id => event[:id] ).delete
end
end
return events
end
######################################################################
# I N S T A N C E M E T H O D S
######################################################################
### Create a new ScheduledEvent object.
###
def initialize( row )
@event = Symphony::Metronome::IntervalExpression.parse( row[:expression], row[:created] )
@options = row.delete( :options )
@id = row.delete( :id )
self.reset_runtime
unless self.class.splay.zero?
splay = Range.new( - self.class.splay, self.class.splay )
@runtime = self.runtime + rand( splay )
end
end
# The parsed interval expression.
attr_reader :event
# The unique ID number of the scheduled event.
attr_reader :id
# The options hash attached to this event.
attr_reader :options
# The exact time that this event will run.
attr_reader :runtime
### Set the datetime that this event should fire next.
###
def reset_runtime
now = Time.now
# Start time is in the future, so it's sufficent to be considered the run time.
#
if self.event.starting >= now
@runtime = self.event.starting
return
end
# Otherwise, the event should already be running (start time has already
# elapsed), so schedule it forward on it's next interval iteration.
#
@runtime = now + self.event.interval
end
### Perform the action attached to the event. Yields the
### deserialized options, the action ID to the supplied block if
### this event is okay to execute.
###
### Automatically remove the event if it has expired.
###
def fire
rv = self.event.fire?
if rv
opts = Yajl.load( self.options )
yield opts, self.id
end
self.delete if rv.nil?
return rv
end
### Permanently remove this event from the database.
###
def delete
self.log.debug "Removing action %p" % [ self.id ]
self.class.db[ :metronome ].filter( :id => self.id ).delete
end
### Comparable interface, order by next run time, soonest first.
###
def <=>( other )
return self.runtime <=> other.runtime
end
end # Symphony::Metronome::ScheduledEvent

156
lib/symphony/metronome/scheduler.rb Normal file
View file

@ -0,0 +1,156 @@
#!/usr/bin/env ruby
# vim: set nosta noet ts=4 sw=4:
require 'symphony'
require 'symphony/metronome'
### Manage the delta queue of events and associated actions.
###
class Symphony::Metronome::Scheduler
extend Loggability, Configurability
include Symphony::SignalHandling
log_to :symphony
config_key :metronome
# Signals the daemon responds to.
SIGNALS = [ :HUP, :INT, :TERM ]
CONFIG_DEFAULTS = {
:listen => true
}
class << self
# Should Metronome register and schedule events via AMQP?
# If +false+, you'll need a separate way to add event actions
# to the database, and manually HUP the daemon.
attr_reader :listen
end
### Configurability API
###
def self::configure( config=nil )
config = self.defaults.merge( config || {} )
@listen = config.delete( :listen )
end
### Create and start an instanced daemon.
###
def self::run( &block )
return new( block )
end
### Actions to perform when creating a new daemon.
###
private_class_method :new
def initialize( block ) #:nodoc:
# Start the queue subscriber for schedule changes.
#
if self.class.listen
Symphony::Metronome::ScheduledEvent.db.disconnect
@child = fork do
$0 = 'Metronome (listener)'
Symphony::Metronome::ScheduleTask.run
end
Process.setpgid( @child, 0 )
end
# Signal handling for the master (this) process.
#
self.set_up_signal_handling
self.set_signal_traps( *SIGNALS )
@queue = Symphony::Metronome::ScheduledEvent.load
@proc = block
# Enter the main loop.
self.start
rescue => err
self.log.error "%p while running: %s" % [ err.class, err.message ]
self.log.debug " " + err.backtrace.join( "\n " )
Process.kill( 'TERM', @child ) if self.class.listen
end
# The sorted set of ScheduledEvent objects.
attr_reader :queue
#########
protected
#########
### Main daemon sleep loop.
###
def start
$0 = "Metronome%s" % [ self.class.listen ? ' (executor)' : '' ]
@running = true
loop do
wait = nil
if ev = self.queue.first
wait = ev.runtime - Time.now
wait = 0 if wait < 0
self.log.info "Next event in %0.3f second(s) (id: %d)..." % [ wait, ev.id ]
else
self.log.warn "No events scheduled. Waiting indefinitely..."
end
self.process_events unless self.wait_for_signals( wait )
break unless @running
end
end
### Dispatch incoming signals to appropriate handlers.
###
def handle_signal( sig )
case sig
when :TERM, :INT
@running = false
Process.kill( sig.to_s, @child ) if self.class.listen
when :HUP
@queue = Symphony::Metronome::ScheduledEvent.load
self.queue.each{|ev| ev.fire(&@proc) if ev.event.recurring }
else
self.log.debug "Unhandled signal: %s" % [ sig ]
end
end
### Process all event that have reached their runtime.
###
def process_events
now = Time.now
self.queue.each do |ev|
next unless now >= ev.runtime
self.queue.delete( ev )
rv = ev.fire( &@proc )
# Reschedule the event and place it back on the queue.
#
if ev.event.recurring
ev.reset_runtime
self.queue.add( ev ) unless rv.nil?
# It was a single run event, torch it!
#
else
ev.delete
end
end
end
end # Symphony::Metronome::Scheduler

81
lib/symphony/tasks/scheduletask.rb Normal file
View file

@ -0,0 +1,81 @@
#!/usr/bin/env ruby
# vim: set nosta noet ts=4 sw=4:
require 'symphony'
require 'symphony/routing'
require 'symphony/metronome'
### Accept metronome scheduling events, translating them
### to DB rows for persistence.
###
class Symphony::Metronome::ScheduleTask < Symphony::Task
include Symphony::Routing
queue_name 'metronome'
timeout 30
### Get a handle to the database.
###
def initialize( * )
@db = Symphony::Metronome::ScheduledEvent.db
@actions = @db[ :metronome ]
super
end
# The Sequel dataset of scheduled event actions.
attr_reader :actions
### Accept a new scheduled event. The payload should be a free
### form hash of options, along with an expression string that
### conforms to IntervalExpression.
###
### {
### :expression => 'run 25 times for an hour',
### :payload => { ... },
### }
###
on 'metronome.create' do |payload, metadata|
raise ArgumentError, 'Invalid payload.' unless payload.is_a?( Hash )
exp = payload.delete( 'expression' )
raise ArgumentError, 'Missing time expression.' unless exp
self.actions.insert(
:created => Time.now,
:expression => exp,
:options => Yajl.dump( payload )
)
self.signal_parent
return true
end
### Delete an existing scheduled event.
### The payload is the id of the action (row) to delete.
###
on 'metronome.delete' do |id, metadata|
self.actions.filter( :id => id.to_i ).delete
self.signal_parent
return true
end
### Tell our parent (the Metronome broadcaster) to re-read its event
### list.
###
def signal_parent
parent = Process.ppid
# Check to make sure we weren't orphaned.
#
if parent == 1
self.log.error "Lost my parent process? Exiting."
exit 1
end
Process.kill( 'HUP', parent )
end
end