Added -window_days setting to Scheduler

[tcl-tlc-base.git] / scripts / scheduler.itcl

# vim: ft=tcl foldmethod=marker foldmarker=<<<,>>> ts=4 shiftwidth=4

# Expected Datasource schema:
#       column 0:       ID
#       column 1:       event_source_script

# event_source_script must define a proc get_events {start_s end_s}
# that returns a list of event descriptions, each a list of
# {epoc_seconds event_data}
# Example (generates hourly events on the hour):
#
# proc get_events {start_s end_s} {
#       set events      {}
#       set s   [expr {($start_s / 3600) * 3600}]
#       while {$s < $end_s} {
#               lappend events  [list $s [list the_posted_event_for $s]]
#               incr s 3600
#       }       
#       return $events
# }

# Handers fired:
#       event_fired(script_id, event_data)      - When a scheduled event fires

class tlc::Scheduler {
        inherit tlc::Baselog tlc::Handlers

        constructor {args} {}
        destructor {}

        public {
                variable events_ds                      ""
                variable script_col                     1
                variable window_days            1.0

                # Test framework helpers
                method _test_pending_events {}
        }

        private {
                variable posted
                variable afterids
                variable refresh_afterid                ""
                variable current_horizon

                method fire {s}
                method process_script {id script start_s end_s}
                method process_event {event}
                method setup_events {}
                method remove_ev_source {id}
                method update_last_run {{s now}}
                method refresh {}
                method get_last_run {}
                method now {}

                method new_item {pool id data}
                method change_item {pool id olddata newdata}
                method remove_item {pool id olddata}

                method new_source {id script}
        }
}


body tlc::Scheduler::constructor {args} { #<<<
        log debug $this
        package require Tcl 8.5

        array set posted        {}
        array set afterids      {}

        configure {*}$args

        foreach reqf {events_ds} {
                if {![info exists $reqf] || [set $reqf] == ""} {
                        error "Must set -$reqf"
                }
        }

        if {![itcl::is object $events_ds] || ![$events_ds isa tlc::Datasource]} {
                error "-events_ds must be a tlc::Datasource or descendant"
        }

        if {[$events_ds isa tlc::DSchan] || [$events_ds isa tlc::Datasource_filter]} {
                log debug "-events_ds is a DSchan or Datasource_filter"
                $events_ds register_handler new_item    [code $this new_item]
                $events_ds register_handler change_item [code $this change_item]
                $events_ds register_handler remove_item [code $this remove_item]
                setup_events
        } else {
                setup_events
        }
}

#>>>
body tlc::Scheduler::destructor {} { #<<<
        log debug $this
        after cancel $refresh_afterid; set refresh_afterid      ""

        foreach {s afterid} [array get afterids] {
                after cancel $afterid
                array unset afterids $s
        }

        if {
                [info exists events_ds] &&
                [itcl::is object $events_ds] &&
                (
                        [$events_ds isa tlc::DSchan] ||
                        [$events_ds isa tlc::Datasource_filter]
                )
        } {
                $events_ds deregister_handler new_item          [code $this new_item]
                $events_ds deregister_handler change_item       [code $this change_item]
                $events_ds deregister_handler remove_item       [code $this remove_item]
                $events_ds deregister_handler init                      [code $this setup_events]
        }
}

#>>>
body tlc::Scheduler::fire {s} { #<<<
        set afterinfo   {}
        foreach id [after info] {
                lappend afterinfo       [list $id [after info $id]]
        }
        log debug "\nposted for this slot: [llength $posted($s)] items: ([join $posted($s) |])\npending afters:\n\t[join $afterinfo \n\t]"
        if {[info exists afterids($s)]} {
                after cancel $afterids($s)
                array unset afterids $s
        }
        if {![info exists posted($s)]} return

        foreach event $posted($s) {
                tlc::try {
                        process_event $event
                } onerr {
                        default {
                                log error "error processing event: $errmsg\n$::errorInfo"
                        }
                }
        }
        array unset posted $s

        update_last_run $s
}

#>>>
body tlc::Scheduler::process_script {id script start_s end_s} { #<<<
        log debug
        set interp      [interp create -safe]

        $interp alias remove_ev_source [code $this remove_ev_source $id]

        tlc::try {
                $interp eval $script
                if {[$interp eval {expr {
                        [info commands get_events] == "get_events"
                }}]} {
                        set events      [$interp eval [list get_events $start_s $end_s]]
                } else {
                        set events      {}
                        log error "Event source $id does not define get_events proc"
                }
        } onerr {
                default {
                        log error "Error initializing safe interpreter with script or calling get_events: $errmsg ($::errorCode)\n$::errorInfo"
                }
        } onok {
                foreach event $events {
                        lassign $event s event_data

                        if {$s < $start_s || $s > $end_s} continue

                        lappend posted($s)      [list $id $event_data]
                        if {![info exists afterids($s)]} {
                                set delta       [expr {$s - [now]}]
                                if {$delta < 1} {
                                        set delta       1
                                }
                                log notice "scheduling after for $delta seconds time (slot $s)"
                                set afterids($s)        [after [expr {$delta * 1000}] \
                                                [code $this fire $s]]
                        }
                }
        }

        interp delete $interp
}

#>>>
body tlc::Scheduler::process_event {event} { #<<<
        log debug
        lassign $event script_id event_data
        invoke_handlers event_fired $script_id $event_data
}

#>>>
body tlc::Scheduler::setup_events {} { #<<<
        log debug
        after cancel $refresh_afterid; set refresh_afterid      ""

        set last_run    [get_last_run]

        set end_s       [expr {[now] + int($window_days * 86400)}]
        set current_horizon     $end_s
        
        set id_column   [$events_ds cget -id_column]
        foreach row [$events_ds get_list {}] {
                set script_id   [lindex $row $id_column]
                new_item foo $script_id $row
        }

        set delta       [expr {($end_s - [now]) * 1000}]

        set refresh_afterid     [after $delta [code $this refresh]]
}

#>>>
body tlc::Scheduler::remove_ev_source {id} { #<<<
        log debug
        set now                         [now]

        foreach {s events} [array get posted] {
                set new_events  {}
                foreach event $events {
                        set ev_script_id        [lindex $event 2]
                        if {$ev_script_id != $id} {
                                lappend new_events $event
                        }
                }

                if {[llength $new_events] != [llength $events]} {
                        if {[llength $new_events] == 0} {
                                if {$s >= $now} {
                                        after cancel $afterid
                                        array unset afterids $s
                                        array unset posted $s
                                }
                        } else {
                                set posted($s)  $new_events
                        }
                }
        }
}

#>>>
body tlc::Scheduler::update_last_run {{s now}} { #<<<
        log debug
        if {$s == "now"} {
                set s   [now]
        }

        set fp  [open "last_run" w]
        puts $fp $s
        close $fp
}

#>>>
body tlc::Scheduler::refresh {} { #<<<
        log debug
        set now         [now]
        update_last_run $now

        foreach {s afterid} {
                if {$s >= $now} {
                        after cancel $afterid
                        array unset afterids $s
                        array unset posted $s
                }
        }
        setup_events
}

#>>>
body tlc::Scheduler::get_last_run {} { #<<<
        log debug
        if {[catch {
                set fp  [open "last_run" r]
        } errmsg]} {
                set last_run    ""
        } else {
                set last_run    [read $fp]
                close $fp
        }

        if {![string is integer -strict $last_run]} {
                set last_run    [now]
        }

        return $last_run
}

#>>>
body tlc::Scheduler::new_item {pool id data} { #<<<
        log debug
        set script      [lindex $data $script_col]
        new_source $id $script
}

#>>>
body tlc::Scheduler::change_item {pool id olddata newdata} { #<<<
        set oldscript   [lindex $olddata 1]
        set newscript   [lindex $newdata 1]
        if {$oldscript == $newscript} return
        log debug
        remove_ev_source $id
        new_item $pool $id $newdata
}

#>>>
body tlc::Scheduler::remove_item {pool id olddata} { #<<<
        log debug
        remove_ev_source $id
}

#>>>
body tlc::Scheduler::new_source {id script} { #<<<
        log debug
        tlc::assert {[info exists current_horizon]} "current_horizon defined"
        set start_s     [now]
        set end_s       $current_horizon

        process_script $id $script $start_s $end_s
}

#>>>
body tlc::Scheduler::now {} { #<<<
        return [clock seconds]
}

#>>>
body tlc::Scheduler::_test_pending_events {} { #<<<
        log debug
        set build       {}
        foreach s [lsort -integer -increasing [array names posted]] {
                lappend build   $s $posted($s)
        }
        return $build
}

#>>>