R/q.monitor.R

In okux/qrmarkdown: Job Queue Management For Rmarkdown

q.monitor <- function( wait=FALSE )
{

  dox <- function()
  {
    if(file.exists(.pkg.log$scheduler.shutdown))
      file.remove(.pkg.log$scheduler.shutdown)

    message('scheduler is started for hourly monitoring')
    n.sleep <- 4
    save(n.sleep, file=.pkg.log$scheduler.file)

    # Core loop monitoring when job need to kick off
    while(!is.scheduler.shutdown())
    {
      qlist <- q.show()
      if(qlist$schedule > 0 )
      {
        TDIR<-sprintf('%s/schedule',.pkg.log$log.dir)

        # matching date and time
        NOW.WDAY <- weekdays(Sys.Date())
        NOW.HR   <- format( Sys.time(), format = '%H')

        sfiles <- dir(TDIR, full.names = TRUE, recursive = TRUE)

        for (item in sfiles)
        {
          load(item)

          if(is.null(ticket$wday) || is.na(ticket$wday))
            T_WCHECK <- TRUE

          if(ticket$wday == NOW.WDAY)
            T_WCHECK <- TRUE

          if(is.null(ticket$hour) || is.na(ticket$hour))
            T_HCHECK <- TRUE

          if(ticket$hour == NOW.HR)
            T_HCHECK <- TRUE

          if(T_HCHECK && T_WCHECK)
          {
            message('queueing job ..\t', basename(item))
            TPATH <- sprintf('%s/inbox/%s',
                             .pkg.log$log.dir,basename(item))

            if(file.exists(TPATH)) # if already exist, refresh file
              file.remove(TPATH)

            file.copy(item,basename(TPATH))

          }

        }# for

        # hour base transaction
        Sys.sleep(60*60)
      }

    }

    message('scheduler shutdown bye')

  }

  if(!wait)
  {
    code="
    require(qrmarkdown)
    q.monitor(wait=TRUE)
    "

    tmpfile <- tempfile()
    sink(file=tmpfile)
    cat(code)
    sink()

    rscript.cmd <- sprintf('Rscript %s',
                           tmpfile)

    system(rscript.cmd,wait = FALSE)

  } else
    dox()
}