0.0.7 improved action call. now allows binding of arguments and passes active schedule as last argument
David Kebler
parent
2f6901e7ae
commit
0a467059e5
|
|
@ -1,36 +0,0 @@
|
|||
import Schedule from '../src/schedule.js'
|
||||
import Runner from '../src/runner.js'
|
||||
|
||||
const HOUR = 0
|
||||
const MINUTE = 0
|
||||
const DELTA = 6
|
||||
const DURATION = 2
|
||||
const DEV = true // if true delta and duration are seconds
|
||||
const NZONES = 4
|
||||
|
||||
let zones = []
|
||||
for (let zone=0; zone < NZONES; zone++) {
|
||||
let opts = {dev:DEV, simulanteous:(zone % 2), name:'zone-'+(zone+1),hour:HOUR,minute:MINUTE,delta:DELTA, duration:DURATION}
|
||||
zones[zone] = new Schedule (opts)
|
||||
// console.log(zones[zone].name,'>hr:min:delta:duration:simultaneous=',
|
||||
// zones[zone].hour,
|
||||
// zones[zone].minute,
|
||||
// zones[zone].delta,
|
||||
// zones[zone].duration,
|
||||
// console.dir(val)
|
||||
// zones[zone].simultanecous)
|
||||
// zones[zone].on('update',val => {
|
||||
// console.log('just updated schedule')
|
||||
// console.dir(val)
|
||||
// })
|
||||
zones[zone].update()
|
||||
}
|
||||
|
||||
const runner = new Runner ({
|
||||
name: 'irrigation',
|
||||
// one: true,
|
||||
schedules: zones,
|
||||
})
|
||||
|
||||
runner.start()
|
||||
setTimeout(() => {runner.stop(true)},10*1000)
|
||||
|
|
@ -0,0 +1,108 @@
|
|||
import Schedule from '../src/schedule.js'
|
||||
import Runner from '../src/runner.js'
|
||||
|
||||
import delay from 'delay'
|
||||
|
||||
const HOUR = 0
|
||||
const MINUTE = 0
|
||||
const DELTA = 5
|
||||
const DURATION = 4
|
||||
const DEV = true // if true delta and duration are seconds
|
||||
const NZONES = 1
|
||||
|
||||
|
||||
// MUST be promise returning and NOT an arrow function
|
||||
function startPromise (zone,activeSch) {
|
||||
console.log('promise return start function', this.evName)
|
||||
return new Promise(resolve => {
|
||||
console.log('starting>>>',zone.id)
|
||||
console.log('---waiting-----', zone.duration,'secs duration to complete')
|
||||
const tick = setInterval(()=>{
|
||||
console.log('duration tick')
|
||||
}, 1000)
|
||||
this.once('abort:'+activeSch.runID,()=> {
|
||||
clearTimeout(run)
|
||||
console.log('aborting run>>>', activeSch.runID)
|
||||
})
|
||||
const run = setTimeout(()=>{
|
||||
console.log('stopping>>>',activeSch.runID)
|
||||
clearInterval(tick)
|
||||
this.removeAllListeners('abort:'+activeSch.runID)
|
||||
resolve()
|
||||
}, zone.duration*1000)
|
||||
})
|
||||
}
|
||||
|
||||
function stop (zone,activeSch) {
|
||||
console.log('aborting run for action id',activeSch.runID, 'zone',zone.id)
|
||||
this.emit('abort:'+activeSch.runID)
|
||||
}
|
||||
|
||||
async function startAsync (zone,activeSch) {
|
||||
// console.log(arguments)
|
||||
console.log('async start function')
|
||||
console.log('-----starting-------',zone.id, activeSch.runID)
|
||||
console.log('---waiting-----', zone.duration,'secs duration to complete')
|
||||
const tick = setInterval(()=>{
|
||||
console.log('duration tick')
|
||||
}, 1000)
|
||||
const duration = delay(zone.duration*1000)
|
||||
this.once('abort:'+activeSch.runID,()=> {
|
||||
console.log('aborting run>>>', activeSch.runID)
|
||||
duration.clear()
|
||||
})
|
||||
await duration
|
||||
clearInterval(tick)
|
||||
this.removeAllListeners('abort:'+activeSch.runID)
|
||||
console.log('stopping', activeSch.runID)
|
||||
}
|
||||
|
||||
|
||||
(async () => {
|
||||
|
||||
let zones = []
|
||||
let schs = []
|
||||
for (let zone=0; zone < NZONES; zone++) {
|
||||
let opts = {dev:DEV, simultaneous:(zone % 2), name:'sch-zone-'+(zone+1),id:'zone-'+(zone+1) ,hour:HOUR,minute:MINUTE,delta:DELTA, duration:DURATION}
|
||||
zones[zone] = opts
|
||||
console.log(zones[zone].id,'>hr:min:delta:duration:simultaneous=',
|
||||
zones[zone].hour,
|
||||
zones[zone].minute,
|
||||
zones[zone].delta,
|
||||
zones[zone].duration,
|
||||
zones[zone].simultaneous)
|
||||
|
||||
schs[zone] = new Schedule (opts)
|
||||
schs[zone].on('update',val => {
|
||||
console.log('just updated schedule')
|
||||
console.dir(val)
|
||||
})
|
||||
schs[zone].update()
|
||||
|
||||
console.log('register action !!!!!!!!!!!!!!!!!!!!!')
|
||||
schs[zone].registerStartAction(startAsync,zones[zone])
|
||||
schs[zone].registerStopAction(stop,zones[zone])
|
||||
console.log('register action done!!!!!!!!!!!!!!!!!!!!!')
|
||||
|
||||
}
|
||||
|
||||
const runner = new Runner ({
|
||||
name: 'irrigation',
|
||||
// one: true,
|
||||
schedules: schs,
|
||||
})
|
||||
|
||||
runner.start()
|
||||
const cd = setInterval(()=>{
|
||||
console.log('runner, next schedule trigger in', runner.countdown)
|
||||
},1000)
|
||||
setTimeout(() => {
|
||||
runner.stop(true)
|
||||
clearInterval(cd)},
|
||||
15*1000
|
||||
)
|
||||
|
||||
})().catch(err => {
|
||||
console.error('FATAL: UNABLE TO START SYSTEM!\n',err)
|
||||
// process.kill(process.pid, 'SIGTERM')
|
||||
})
|
||||
|
|
@ -1,25 +1,52 @@
|
|||
import Schedule from '../src/schedule.js'
|
||||
// import Queue from 'src/queue.js'
|
||||
|
||||
const zone1 = new Schedule ({
|
||||
name: 'zone1',
|
||||
|
||||
function demoStartAction (duration) {
|
||||
return new Promise(resolve => {
|
||||
console.log('starting schedule',this.name)
|
||||
let stopTO
|
||||
this.once('abort', () => {
|
||||
console.log('action was aborted')
|
||||
resolve('action aborted')
|
||||
clearTimeout(stopTO)
|
||||
})
|
||||
console.log('waiting',duration,'secs duration to complete')
|
||||
stopTO = setTimeout(()=>{
|
||||
console.log('action for',this.name, 'has ended')
|
||||
resolve('normal completion')
|
||||
this.removeAllListeners('abort')
|
||||
}, duration*1000)
|
||||
})
|
||||
}
|
||||
|
||||
function demoStopAction () {
|
||||
console.log('forced stop action for', this.name)
|
||||
this.emit('abort')
|
||||
}
|
||||
|
||||
const sch = new Schedule ({
|
||||
name: 'demo',
|
||||
dev:true,
|
||||
hour: 2,
|
||||
minute: 0,
|
||||
delta: 6,
|
||||
duration: 1
|
||||
})
|
||||
|
||||
console.log('zone1>hr:min:delta:duration=', zone1.hour,zone1.minute,zone1.delta,zone1.duration)
|
||||
|
||||
zone1.on('update',val => {
|
||||
sch.on('update',val => {
|
||||
console.dir(val)
|
||||
})
|
||||
|
||||
zone1.update()
|
||||
zone1.minute=30
|
||||
zone1.hour=6
|
||||
zone1.delta=12
|
||||
sch.update()
|
||||
const DURATION = 5
|
||||
sch.registerStartAction(demoStartAction,DURATION)
|
||||
sch.registerStopAction(demoStopAction)
|
||||
|
||||
sch.startAction()
|
||||
|
||||
setTimeout(()=>{
|
||||
zone1.update()
|
||||
process.exit()
|
||||
},6000)
|
||||
sch.startAction()
|
||||
},DURATION*1000)
|
||||
|
||||
setTimeout(()=>{
|
||||
sch.stopAction() // abort the run
|
||||
},(DURATION*2-2)*1000)
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
{
|
||||
"name": "@uci-utils/scheduler",
|
||||
"version": "0.0.6",
|
||||
"version": "0.0.7",
|
||||
"description": "an interval reoccuring style scheduler and runner",
|
||||
"main": "src/index.js",
|
||||
"scripts": {
|
||||
"example:sch": "nodemon -r esm examples/schedule.js",
|
||||
"example:que": "nodemon -r esm examples/queue.js"
|
||||
"example:run": "nodemon -r esm examples/runner.js"
|
||||
},
|
||||
"author": "",
|
||||
"license": "ISC",
|
||||
|
|
@ -16,6 +16,7 @@
|
|||
"moment-duration-format": "^2.3.2"
|
||||
},
|
||||
"devDependencies": {
|
||||
"delay": "^4.3.0",
|
||||
"esm": "^3.2.25",
|
||||
"nodemon": "^1.19.1"
|
||||
}
|
||||
|
|
|
|||
106
src/runner.js
106
src/runner.js
|
|
@ -31,32 +31,41 @@ class UCIScheduleRunner extends EventEmitter {
|
|||
return (this.schedules[0]||{}).name
|
||||
}
|
||||
|
||||
get nextTS() {
|
||||
return (this.schedules[0]||{}).nextTS
|
||||
}
|
||||
|
||||
get nextDT(){
|
||||
return (this.schedules[0]||{}).nextDT
|
||||
}
|
||||
|
||||
|
||||
init() {
|
||||
|
||||
this.on('active', active=>{
|
||||
this._delayedListener = this.on('active', active => {
|
||||
if (!active && this._delayed.length) {
|
||||
const sch=this._delayed.shift()
|
||||
// console.log('a delayed action now proceeding====>', sch.name, sch.toid)
|
||||
start.call(this,sch)
|
||||
console.log('a delayed action now proceeding====>', sch.name, sch.toid)
|
||||
this.startScheduleAction(sch)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// main timeout based on the first schedule to trigger
|
||||
setTimeout () {
|
||||
// clearTimeout(this._timeout)
|
||||
const next = this.schedules[0] || {}
|
||||
this._toID++
|
||||
console.log('+++++++++++++++++++++ next timeout',this._toID, 'in', Math.round(next.countdownMS/1000),'++++++++++++++++++++++++++')
|
||||
this._timeout = setTimeout(() => {
|
||||
console.log('**********************timeout triggered', this._toID, '********************************')
|
||||
let dur = 0
|
||||
// let dur = 0
|
||||
// don't mutate original queue, make copy for active list
|
||||
const queue = clone(this.schedules)
|
||||
this._active[this._toID]=[]
|
||||
queue.forEach(sch =>{ // add first and any others set for same time
|
||||
dur += sch.durationMS
|
||||
// dur += sch.durationMS
|
||||
// console.log('check if active', sch.name, sch.countdownMS, dur)
|
||||
if (sch.countdownMS <= dur) this._active[this._toID].push(sch)
|
||||
if (sch.countdownMS <= 1000) this._active[this._toID].push(sch)
|
||||
if (sch.one) this.removeSchedule(sch.id)
|
||||
})
|
||||
// this._active[this._toID].next = 0
|
||||
|
|
@ -70,13 +79,17 @@ class UCIScheduleRunner extends EventEmitter {
|
|||
console.log('---------------------stopping runner------------------------')
|
||||
clearTimeout(this._timeout)
|
||||
if (now) {
|
||||
console.log('also stopping all in progress schedules')
|
||||
console.log('clearing all delayed start schedules', this._delayed.map(sch=>sch.name))
|
||||
this._delayed = []
|
||||
console.log('stopping all in progress schedules')
|
||||
for (let list in this._active) {
|
||||
this._active[list].forEach(sch=>{
|
||||
clearTimeout(sch._stopTimeout)
|
||||
sch.stopAction()}
|
||||
console.log('active schedule',sch.name)
|
||||
if (sch.stopAction) sch.stopAction()
|
||||
else console.log(sch.name,'no stop action registered can not force active schedule to stop')}
|
||||
)
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -94,27 +107,19 @@ class UCIScheduleRunner extends EventEmitter {
|
|||
const idx = this._active[toid].length-count
|
||||
const sch = this._active[toid][idx]
|
||||
sch.toid = toid
|
||||
// let next = this._active[toid].next
|
||||
// console.log(idx,sch)
|
||||
sch.runID = `${toid}:${sch.id}`
|
||||
count--
|
||||
if (idx===0 && !this._activeCount) start.call(this,sch)
|
||||
if (sch.simultaneous) start.call(this,sch)
|
||||
if (sch.simultaneous) this.startScheduleAction(sch)
|
||||
else if (!this._activeCount) this.startScheduleAction(sch)
|
||||
// if (idx===0 && !this._activeCount) this.startScheduleAction(sch)
|
||||
else {
|
||||
// console.log('adding schedule to delay',sch.name,sch.toid)
|
||||
console.log('adding schedule to delay',sch.name,sch.toid)
|
||||
this._delayed.push(sch)
|
||||
// console.log(this._delayed.length)
|
||||
console.log(this._delayed.length)
|
||||
}
|
||||
this.trigger(count,toid)
|
||||
}
|
||||
|
||||
get nextTS() {
|
||||
return (this.schedules[0]||{}).nextTS
|
||||
}
|
||||
|
||||
get nextDT(){
|
||||
return (this.schedules[0]||{}).nextDT
|
||||
}
|
||||
|
||||
// schedule instance
|
||||
addSchedule (sch) {
|
||||
if (getBaseClass(sch) === 'UCISchedule') {
|
||||
|
|
@ -140,6 +145,34 @@ class UCIScheduleRunner extends EventEmitter {
|
|||
this.getSchedule(id).update()
|
||||
}
|
||||
|
||||
async startScheduleAction (sch, toid) {
|
||||
toid = sch.toid || toid
|
||||
console.log('----',sch.runID,'scheduled run started-----')
|
||||
// TODO check if async or promise
|
||||
this._activeCount++
|
||||
this.emit('active',this._activeCount) // runner has active schedules
|
||||
sch.startAction(sch.runID)
|
||||
sch.once(sch.runID,removeScheduleFromActiveList.bind(this)) // emitted on abort too
|
||||
|
||||
function removeScheduleFromActiveList() {
|
||||
console.log('-----',sch.runID,'run complete or aborted removing from active list ----')
|
||||
this._activeCount--
|
||||
// done remove from active list
|
||||
this._active[toid]=this._active[toid].filter(s=>s.id!==sch.id)
|
||||
if (!this._active[toid].length) {
|
||||
// none left on that list, remove it
|
||||
console.log('===========actions for timeout', toid, 'are complete===========')
|
||||
delete this._active[toid]
|
||||
}
|
||||
// console.log('schedule action complete', sch.name,toid, this._activeCount)
|
||||
this.emit('active',this._activeCount) // runner has active schedules ??
|
||||
}
|
||||
}
|
||||
|
||||
disableSchedule(id) {
|
||||
this.enableSchedule(id,'false')
|
||||
}
|
||||
|
||||
enableSchedule(id,state) {
|
||||
if (state == null) state = true
|
||||
if (state) {
|
||||
|
|
@ -171,33 +204,12 @@ class UCIScheduleRunner extends EventEmitter {
|
|||
sort () {
|
||||
this.schedules.sort((a,b) => a.nextTS - b.nextTS)
|
||||
}
|
||||
}
|
||||
|
||||
} // end class
|
||||
|
||||
export default UCIScheduleRunner
|
||||
export { UCIScheduleRunner as Runner }
|
||||
|
||||
function start (sch, toid) {
|
||||
toid = sch.toid || toid
|
||||
console.log('-----timeoutid',toid, 'start action-----')
|
||||
sch.startAction()
|
||||
this._activeCount++
|
||||
this.emit('active',1)
|
||||
sch._stopTimeout = setTimeout(() =>{ // stop action timeout
|
||||
console.log('---timeoutid',toid, 'stop action----')
|
||||
sch.stopAction()
|
||||
this._activeCount--
|
||||
// done remove from active list
|
||||
this._active[toid]=this._active[toid].filter(s=>s.id!==sch.id)
|
||||
if (!this._active[toid].length) {
|
||||
// none left on that list, remove it
|
||||
console.log('===========actions for timeout', toid, 'are complete===========')
|
||||
delete this._active[toid]
|
||||
}
|
||||
// console.log('schedule action complete', sch.name,toid, this._activeCount)
|
||||
this.emit('active',this._activeCount)
|
||||
},sch.durationMS)
|
||||
}
|
||||
|
||||
function getBaseClass(targetClass){
|
||||
|
||||
if(targetClass instanceof Object){
|
||||
|
|
|
|||
|
|
@ -16,17 +16,15 @@ class UCISchedule extends EventEmitter {
|
|||
this.dev = opts.dev
|
||||
this.minute = opts.minute || 0
|
||||
this.delta = opts.delta || 6 // time to next trigger in hours
|
||||
this.duration = opts.duration || 10 // in minutes
|
||||
// computed values
|
||||
this.nextTS = 0 // the next trigger time in seconds
|
||||
this.enabled = opts.enabled || true
|
||||
this.enabled = opts.enabled==null ? true : opts.enabled
|
||||
this.active = false
|
||||
// this.active = false // if schedule is currently active
|
||||
// this.on('active',active=>this.active=active)
|
||||
this.simultaneous = opts.simulanteous // if true delay this scheduled event until current scheduled event completes
|
||||
this.simultaneous = opts.simultaneous // if true = run even if there are other active schedules
|
||||
// this.lastActiveTS
|
||||
this._startAction = opts.startAction || defaultStartAction.bind(this) // single (or array) of function(s)
|
||||
this._stopAction = opts.stopAction || defaultStopAction.bind(this) // single (or array) of function(s)
|
||||
// Actions MUST be promise returning or async
|
||||
this._startAction = opts.startAction // || defaultStartAction.bind(this) // single (or array) of function(s)
|
||||
this._stopAction = opts.stopAction // || defaultStopAction.bind(this) // single (or array) of function(s)
|
||||
this.update()
|
||||
}
|
||||
|
||||
|
|
@ -34,29 +32,12 @@ class UCISchedule extends EventEmitter {
|
|||
return moment.duration(parseInt((this.nextTS - Math.floor(Date.now()/1000))),'seconds').format('hh:mm:ss')
|
||||
}
|
||||
|
||||
// get countdownActive() {
|
||||
// if (this.active) return moment.duration(parseInt((this.duration*60 + this.nextTS - Math.floor(Date.now()/1000))),'seconds').format('hh:mm:ss')
|
||||
//
|
||||
// }
|
||||
|
||||
get countdownMS() {
|
||||
return (this.nextTS*1000 - Date.now())
|
||||
}
|
||||
|
||||
get durationMS() {
|
||||
return (this.duration*(this.dev ? 1 :60)*1000)
|
||||
}
|
||||
|
||||
get nextDT() { return moment(this.nextTS*1000).format('ddd, MMM Do h:mm A') }
|
||||
|
||||
// get activeDuration() {
|
||||
// return moment.duration((Math.floor(Date.now()/1000) - this.nextTS),'seconds').format('hh:mm:ss')
|
||||
// }
|
||||
|
||||
// get activeRemaining() {
|
||||
// return moment.duration((this.nextTS - Math.floor(Date.now()/1000) + this.duration * 60),'seconds').format('hh:mm:ss')
|
||||
// }
|
||||
|
||||
update() { // all TS in seconds
|
||||
// console.log('updating',this.hour,this.minute,this.delta, this.dev)
|
||||
let baseTS = this.hour*3600+this.minute*60
|
||||
|
|
@ -74,39 +55,48 @@ class UCISchedule extends EventEmitter {
|
|||
this.emit('update',{id:this.id, ts:this.nextTS,dt:this.nextDT,countdown:this.countdown,simultaneous:this.simultaneous})
|
||||
}
|
||||
|
||||
// MUST! be async or promise TODO make that check
|
||||
registerStartAction(func) {
|
||||
//
|
||||
// if (!Array.isArray(func)) func = [func]
|
||||
this._startAction = func
|
||||
if (typeof func!=='function') throw new Error('start action must be a function and be async or promise returning')
|
||||
const args = [...arguments]
|
||||
args.shift()
|
||||
console.log('registering start action with these arguments to be passed\n',args)
|
||||
this._startAction = func.bind(this,...args)
|
||||
}
|
||||
|
||||
registerStopAction(func) {
|
||||
// if (!Array.isArray(func)) func = [func]
|
||||
this._stopAction = func
|
||||
if (typeof func!=='function') throw new Error('stop action must be a function')
|
||||
const args = [...arguments]
|
||||
args.shift()
|
||||
console.log('registering stop action with these arguments to be passed\n',args)
|
||||
this._stopAction = func.bind(this,...args)
|
||||
}
|
||||
// todo support a array of functions
|
||||
async startAction (data) {
|
||||
|
||||
async startAction () {
|
||||
if (this._stopAction) {
|
||||
this.active = true
|
||||
this.on('active',true)
|
||||
this._startAction(data)
|
||||
}
|
||||
|
||||
async stopAction (data) {
|
||||
this.emit('active',true)
|
||||
// console.log('passed run id name, prop run id',ev, this.runID)
|
||||
// console.log(this)
|
||||
await this._startAction(this)
|
||||
this.active = false
|
||||
this.on('active',false)
|
||||
this._stopAction(data)
|
||||
this.emit('active',false)
|
||||
this.emit(this.runID) // runner listens for this ID of active schedule to know when it's done
|
||||
} else console.log('no registered start action, register a start action!')
|
||||
}
|
||||
|
||||
async stopAction () {
|
||||
if (this._stopAction) {
|
||||
await this._stopAction(this)
|
||||
this.emit(this.runID)
|
||||
this.active = false
|
||||
this.emit('active',false)
|
||||
}
|
||||
else console.log('no registered stop action, unable to abort, action will complete normally')
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
function defaultStartAction () {
|
||||
console.log('start action for', this.name,moment(Date.now()).format('ddd, MMM Do h:mm:ss A'))
|
||||
}
|
||||
|
||||
function defaultStopAction () {
|
||||
console.log('stop action for', this.name,moment(Date.now()).format('ddd, MMM Do h:mm:ss A'))
|
||||
|
||||
}
|
||||
|
||||
export default UCISchedule
|
||||
export {UCISchedule as Schedule}
|
||||
|
|
|
|||
Loading…
Reference in New Issue