R/queue-rds.R

In dipsaus: A Dipping Sauce for Data Analysis and Visualizations

FileQueue <- R6::R6Class(
  classname = 'FileQueue',
  inherit = AbstractQueue,
  private = list(
    root_path = character(0),
    head_file = character(0),
    total_file = character(0),
    header_file = character(0),
    db_dir = character(0)
  ),
  public = list(
    # Get head so that we know where we are in the queue
    `@get_head` = function(){
      scan(file = private$head_file, what = integer(), n = 1, quiet = TRUE)
    },
    `@set_head` = function(v){
      write(v, private$head_file, ncolumns = 1, append = FALSE)
    },

    # Get total number of items in the queue
    `@get_total` = function(){
      scan(file = private$total_file, what = integer(), n = 1, quiet = TRUE)
    },
    `@set_total` = function(v){
      write(v, private$total_file, ncolumns = 1, append = FALSE)
    },

    `@inc_total` = function(n=1){
      con <- file(description = private$total_file, open = 'r+')
      on.exit(close(con))
      total <- scan(file = con, what = integer(), n = 1, quiet = TRUE) + n
      write(total, file = con, ncolumns = 1, append = FALSE)
      invisible(total)
    },

    `@append_header` = function(msg, ...){
      con <- file(private$header_file, 'a+')
      on.exit(close(con))
      writeLines(msg, con)
      return(length(msg))
    },

    `@store_value` = function(value, key){
      path <- file.path(private$db_dir, key)
      saveRDS(value, file = path)
      key
    },
    restore_value = function(hash, key, preserve = FALSE){
      path <- file.path(private$db_dir, key)
      re <- tryCatch({
        readRDS(path)
      }, error = function(e){
        NULL
      })
      if(!preserve){
        unlink(path)
      }
      re
    },

    `@log` = function(n = -1, all = FALSE){
      if( all ){ head <- 0 }else{ head <- self$head }
      total <- self$total
      count <- total - head
      if( n <= 0 ){ n <- count }else{ n <- min(n, count) }
      if( n == 0 ){ return() }
      re <- read.table(private$header_file, skip = head, header = TRUE, nrows = n)
      stringr::str_split_fixed(re[[1]], '\\|', 4)
    },

    `@reset` = function() {
      self$`@set_head`(0)
      self$`@set_total`(0)
      write('HEADER', file = private$header_file, append = FALSE)
      fs <- list.files(private$db_dir, full.names = TRUE)
      lapply(fs, unlink)
    },

    `@clean` = function(preserve = FALSE, ...) {
      head <- self$head
      total <- self$total
      if( total - head < 1 ){
        self$`@reset`()
      }else{
        re <- readLines(private$header_file)
        if( !preserve && head > 0 ){
          lapply(seq_len(head), function(ii){
            re <- self$print_item(stringr::str_split_fixed(re[[ii+1]], '\\|', n = 4))
            unlink(file.path(private$db_dir, re$hash))
          })
        }
        writeLines(re[c(1, (head+2) : (total+1))], private$header_file)
        self$head <- 0
        self$total <- total - head
      }
    },
    `@validate` = function() {
      assert_dir(private$root_path)
      assert_dir(private$db_dir)
      assert_file_scalar(private$head_file)
      assert_file_scalar(private$total_file)
      assert_file(private$header_file)
    },

    `@connect` = function(path){
      if(!missing(path)){
        path <- normalizePath(path, mustWork = TRUE)
        private$root_path <- path
      }

      if( !dir.exists( private$root_path ) ){
        cat2('path is not a directory', level = 'FATAL')
      }

      tmp <- file.path(private$root_path, 'HEAD')
      file_create(tmp)
      private$head_file <- normalizePath(tmp, mustWork = TRUE)

      head <- scan(tmp, what = integer(), n = 1, nlines = 1, quiet = TRUE)
      if(length(head) != 1 || !is.numeric(head) || head < 0){
        self$`@set_head`(0)
      }


      tmp <- file.path(private$root_path, 'TOTAL')
      file_create(tmp)
      private$total_file <- normalizePath(tmp, mustWork = TRUE)

      total <- scan(tmp, what = integer(), n = 1, nlines = 1, quiet = TRUE)
      if(length(total) != 1 || !is.numeric(total) || total < 0){
        self$`@set_total`(0)
      }

      tmp <- file.path(private$root_path, 'KEYS')
      if( !file.exists(tmp) ){
        file_create(tmp)
        writeLines('HEADER', tmp)
      }
      private$header_file <- normalizePath(tmp, mustWork = TRUE)


      tmp <- file.path(private$root_path, 'DB')
      dir_create(tmp)
      private$db_dir <- normalizePath(tmp, mustWork = TRUE)


      self$`@clean`(preserve = TRUE)
      self$`@validate`()

    },

    initialize = function(path = tempfile()){
      dir_create(path)
      private$root_path <- normalizePath(path)
      lockpath <- file.path(path, 'LOCK')
      if(!file.exists(lockpath)){
        writeLines(rand_string(), lockpath)
      }
      self$lockfile <- readLines(con = lockpath, n = 1)
      self$connect()
    },

    destroy = function(){
      locker_key(self$lockfile, set_default = FALSE, unset = TRUE)
      dipsaus_unlock(self$lockfile)
      unlink(private$db_dir, recursive = TRUE, force = TRUE)
      unlink(file.path(private$root_path, 'LOCK'), force = TRUE)
      unlink(private$head_file, force = TRUE)
      unlink(private$total_file, force = TRUE)
      unlink(private$header_file, force = TRUE)

      # If user accidentally set root_path to be '/', removing it would be a disaster
      unlink(private$root_path, recursive = TRUE, force = FALSE)

      delayedAssign('.lockfile', {cat2('Queue destroyed', level = 'FATAL')}, assign.env = private)

      invisible()
    }
  )
)