R/uploadServer.R

In jbrzusto/motusServer: Operate a server to process data for http://motus.org

#' watch for and enqueue uploaded files
#'
#' Watch for new files in \code{MOTUS_PATH$UPLOADS};
#' and when a new file is found, queue a job to process it.
#'
#' @param tracing logical scalar, default FALSE.  If TRUE, enter
#'     debug browser before handling each new file upload.
#'
#' @param fileEvent character scalar; default: "CREATE".  The
#' last filesystem event generated by the external program which
#' is placing files in the \code{MOTUS_PATH$UPLOADS} folder.
#' For other upload systems, the appropriate event might be "CLOSE_WRITE"
#' or "MOVED_TO".  The event should only occur once per uploaded file
#' and only after the full file contents are available.
#'
#' @return This function does not return; it is meant for use in an R
#'     script run in the background.  After each subjob is handled,
#'     the function checks for the existence of a file called
#'     \code{MOTUS_PATH$UPLOADS/kill}.  If that file is found,
#'     the function calls quit(save="no").  The file will also
#'     be detected within the call to feed() when the queue
#'     is empty, because it is located in the watched folder.
#'
#' @note this depends on some other process placing uploaded files
#'     into the folder \code{MOTUS_PATH$UPLOADS}.  We're using a
#'     patched version of \link{http://www.projectsend.org} whose last
#'     task for an uploaded file is to create a symlink to it in
#'     \code{MOTUS_PATH$UPLOADS}, and this generates a "CREATE" event.
#'
#' @export
#'
#' @author John Brzustowski \email{jbrzusto@@REMOVE_THIS_PART_fastmail.fm}

uploadServer = function(tracing = FALSE, fileEvent="CREATE") {
    if(tracing)
        options(error=recover)

    ensureServerDirs()
    motusLog("Upload server started")

    ## load jobs
    loadJobs()

    ## get a feed of uploads

    feed = getFeeder(MOTUS_PATH$UPLOADS, messages = fileEvent, tracing=tracing)

    ## kill off the inotifywait process when we exit this function
    on.exit(feed(TRUE), add=TRUE)

    ## the kill file; must be in the same folder as passed to getFeeder,
    ## so that creating the killFile causes a return from feed() below.
    ## Note: files uploaded by users are prepended with USERNAME:TIMESTAMP:
    ## so users can't kill this server by uploading a file called "kill".

    killFile = file.path(MOTUS_PATH$UPLOADS, "kill")

    ## ensure MotusDB is safeSQL connection to motus transfer DB
    openMotusDB()

    repeat {
        upfile = feed()    ## this might might wait a long time

        if (upfile == killFile)
            break

        if (tracing)
            browser()

        ## the files in /sgm/uploads are symlinks to uploaded files in
        ## ProjectSend's upload/files folder.
        ## They have filenames of this form:
        ##    USER:YYYY-mm-ddTHH-MM-SS:filename
        ## where USER is the authenticated userID of the uploader, and the string between ':' is the upload timestamp.

        ## change ownership of the file using a sudo-able script
        ##        safeSys("sudo", "/sgm/bin/chown_sg_www-data.sh", upfile)

        bname = basename(upfile)
        parts = strsplit(bname, ":", fixed=TRUE)[[1]]

        motusProjectID = MotusDB("select motus_project_ID from data_uploads.sg_files where url = %s", file.path(parts[1], bname))

        ## create and enqueue a new upload job
        j = newJob("uploadFile", .parentPath=MOTUS_PATH$INCOMING, motusUserID=as.integer(parts[1]), motusProjectID = motusProjectID, .enqueue=FALSE, filename=bname)

        ## record receipt within the job's log
        jobLog(j, paste("File uploaded:", parts[3]), summary=TRUE)

        jpath = file.path(jobPath(j), "upload")
        dir.create(jpath)
        ## move file to the job's dir

        file.rename(upfile, file.path(jpath, bname))

        ## move the job to the main queue

        j$queue = "0"
        moveJob(j, MOTUS_PATH$QUEUE0)

        cat("Job", unclass(j), "has been entered into queue 0\n")

        if (file.exists(killFile))
            break
    }
    motusLog("Upload server stopped")
    quit(save="no")
}