jobqueue updates

Author: dansmith01

DESCRIPTION

@@ -1,8 +1,8 @@
 Package: webqueue
 Type: Package
 Title: Multicore HTTP Server
-Version: 1.1.0
-Date: 2025-04-04
+Version: 1.2.0
+Date: 2025-05-21
 Authors@R: c(
     person(
       "Daniel P.", "Smith",
@@ -27,14 +27,14 @@ Depends: R (>= 4.2.0)
 Imports: 
     cli,
     httpuv,
+    interprocess (>= 1.2.0),
     jsonlite,
-    jobqueue (>= 1.6.0),
+    jobqueue (>= 1.6.6),
     later,
     parallelly,
     promises,
     R6,
     rlang,
-    semaphore,
     webutils,
     utils
 Suggests: 
@@ -44,3 +44,4 @@ Suggests:
     rmarkdown,
     testthat (>= 3.0.0),
     withr
+Remotes:cmmr/jobqueue

NAMESPACE

@@ -2,22 +2,22 @@
 
 importFrom("cli", "ansi_strip", "cli_abort", "cli_text", "style_italic", "col_grey", "style_bold")
 importFrom("httpuv", "randomPort", "startServer", "staticPath", "staticPathOptions")
-importFrom("jobqueue", "Queue", "Worker", "Job", "%<>%", "%>%")
+importFrom("jobqueue", "jobqueue", "%<>%", "%>%")
 importFrom("jsonlite", "fromJSON", "minify", "prettify", "toJSON", "unbox")
 importFrom("later", "later", "run_now")
 importFrom("parallelly", "availableCores")
 importFrom("promises", "as.promise", "then")
 importFrom("R6", "R6Class")
 importFrom("rlang", "%||%", "as_function", "catch_cnd", "cnd_signal", "env_unbind", "is_formula", "is_environment")
-importFrom("semaphore", "create_semaphore", "decrement_semaphore", "remove_semaphore")
+importFrom("interprocess", "semaphore")
 importFrom("webutils", "parse_http", "parse_query")
 importFrom("utils", "capture.output", "hasName", "str")
 
 export(`cookie`)
 export(`header`)
 export(`js_obj`)
 export(`response`)
-export(`WebQueue`)
+export(`webqueue`)
 
 S3method(print, wq_header)
 S3method(print, wq_response)

R/demo.r

@@ -161,7 +161,7 @@ demo <- function () { # nocov start
   }
 
 
-  svr <- WebQueue$new(
+  svr <- webqueue(
     handler = handler,
     host    = '0.0.0.0',
     port    = 8080L,

R/response.r

@@ -1,7 +1,7 @@
 
 #' Compile an HTTP response.
 #' 
-#' If your WebQueue's `handler` function returns a list, json object, character
+#' If your webqueue's `handler` function returns a list, json object, character
 #' vector, or scalar integer, `response()` will be used to transform that 
 #' result into an HTTP response.\cr\cr
 #' You may also call `response()` within your handler to better customize the 

R/utils.r

@@ -44,6 +44,14 @@ valid_string <- function (x, ok = '', no = '', null_ok = FALSE) {
 
 }
 
+is_cran_check <- function() {
+  if (identical(Sys.getenv("NOT_CRAN"), "true")) {
+    FALSE
+  } else {
+    Sys.getenv("_R_CHECK_PACKAGE_NAME_", "") != ""
+  }
+}
+
 
 code_to_msg <- list(
   '100' = "Continue",

R/webqueue.r

@@ -1,8 +1,6 @@
 
 #' Queues and Services HTTP Requests
 #'
-#' @name WebQueue
-#'
 #' @description
 #' 
 #' Connects the 'httpuv' and 'jobqueue' R packages.
@@ -89,44 +87,110 @@
 #'        static paths. If not set or NULL, then it will use the result from 
 #'        calling `httpuv::staticPathOptions()` with no arguments.
 #' 
+#' 
+#' @return
+#' A `webqueue` object with the following methods:
+#' * `$url`
+#'   - Returns the URL where the server is available.
+#' * `$stop(reason = 'server stopped')`
+#'   - Shuts down the webqueue and all associated subprocesses. Stopped Jobs 
+#'     will have their `$output` set to a object of class `<interrupt/condition>`.
+#'   - `reason` - A brief message for the condition object.
+#'   - Returns this webqueue, invisibly.
 #'
 #' @export
-#' @examples
+#' @examplesIf ! jobqueue:::is_cran_check()
 #'     
 #'     library(webqueue)
 #'     
-#'     wq <- WebQueue$new(function (req) 'Hello World!\n')
+#'     wq <- webqueue(function (req) 'Hello World!\n')
 #'     readLines(wq$url)
 #'     wq$stop()
 #' 
 
-WebQueue <- R6Class(
-  classname = "WebQueue",
+webqueue <- function (
+    handler,
+    host              = '0.0.0.0',
+    port              = 8080L,
+    parse             = NULL,
+    globals           = list(),
+    packages          = NULL,
+    namespace         = NULL,
+    init              = NULL,
+    max_cpus          = availableCores(),
+    workers           = ceiling(max_cpus * 1.2),
+    timeout           = NULL,
+    hooks             = NULL,
+    reformat          = NULL,
+    stop_id           = NULL,
+    copy_id           = NULL,
+    bg                = TRUE,
+    quiet             = FALSE,
+    onHeaders         = NULL,
+    staticPaths       = NULL,
+    staticPathOptions = NULL ) {
+  
+  # Capture curly-brace expression
+  init_subst <- substitute(init)
+  if (isa(init_subst, '{')) init <- init_subst
+  
+  # Forward arguments onto class constructor
+  webqueue_class$new(
+    handler           = handler,
+    host              = host,
+    port              = port,
+    parse             = parse,
+    globals           = globals,
+    packages          = packages,
+    namespace         = namespace,
+    init              = init,
+    max_cpus          = max_cpus,
+    workers           = workers,
+    timeout           = timeout,
+    hooks             = hooks,
+    reformat          = reformat,
+    stop_id           = stop_id,
+    copy_id           = copy_id,
+    bg                = bg,
+    quiet             = quiet,
+    onHeaders         = onHeaders,
+    staticPaths       = staticPaths,
+    staticPathOptions = staticPathOptions )
+  
+}
+
+
+
+#' @noRd
+#' @keywords internal
+
+webqueue_class <- R6Class(
+  classname = "webqueue",
   cloneable = FALSE,
 
   public = list(
 
     #' @description
-    #' Creates an `httpuv::WebServer` with requests handled by a `jobqueue::Queue`.
+    #' Creates an `httpuv::WebServer` with requests handled by a `jobqueue::jobqueue`.
     #'
     #' @return A `WebQueue` object.
     initialize = function (
         handler,
-        host      = '0.0.0.0',
-        port      = 8080L,
-        parse     = NULL,
-        globals   = list(),
-        packages  = NULL,
-        namespace = NULL,
-        init      = NULL,
-        max_cpus  = availableCores(),
-        workers   = ceiling(max_cpus * 1.2),
-        timeout   = NULL,
-        hooks     = NULL,
-        reformat  = NULL,
-        stop_id   = NULL,
-        copy_id   = NULL,
-        bg        = TRUE,
+        host              = '0.0.0.0',
+        port              = 8080L,
+        parse             = NULL,
+        globals           = list(),
+        packages          = NULL,
+        namespace         = NULL,
+        init              = NULL,
+        max_cpus          = availableCores(),
+        workers           = ceiling(max_cpus * 1.2),
+        timeout           = NULL,
+        hooks             = NULL,
+        reformat          = NULL,
+        stop_id           = NULL,
+        copy_id           = NULL,
+        bg                = TRUE,
         quiet             = FALSE,
         onHeaders         = NULL,
         staticPaths       = NULL,
@@ -163,38 +227,38 @@ WebQueue <- R6Class(
       # Launch WebQueue on a different R process
       if (isTRUE(bg)) {
 
-        worker  <- jobqueue::Worker$new()
-        sem     <- create_semaphore()
+        worker  <- jobqueue::worker_class$new()
+        sem     <- interprocess::semaphore()
         start_t <- Sys.time()
 
-        job <- jobqueue::Job$new(
+        job <- jobqueue::job_class$new(
           vars = environment(), 
           expr = { # nocov start
 
             # signals an error if unable to start
-            webqueue::WebQueue$new(
-              handler   = handler,
-              host      = host,
-              port      = port,
-              parse     = parse,
-              globals   = globals,
-              packages  = packages,
-              namespace = namespace,
-              init      = init,
-              max_cpus  = max_cpus,
-              workers   = workers,
-              timeout   = timeout,
-              hooks     = hooks,
-              reformat  = reformat,
-              stop_id   = stop_id,
-              copy_id   = copy_id,
-              bg        = FALSE,
+            webqueue::webqueue(
+              handler           = handler,
+              host              = host,
+              port              = port,
+              parse             = parse,
+              globals           = globals,
+              packages          = packages,
+              namespace         = namespace,
+              init              = init,
+              max_cpus          = max_cpus,
+              workers           = workers,
+              timeout           = timeout,
+              hooks             = hooks,
+              reformat          = reformat,
+              stop_id           = stop_id,
+              copy_id           = copy_id,
+              bg                = FALSE,
               quiet             = quiet,
               onHeaders         = onHeaders,
               staticPaths       = staticPaths,
               staticPathOptions = staticPathOptions )
 
-            semaphore::increment_semaphore(sem)
+            sem$post()
 
             httpuv::service(timeoutMs = Inf)
 
@@ -206,7 +270,7 @@ WebQueue <- R6Class(
 
         cnd <- catch_cnd({
 
-          while (!decrement_semaphore(sem, wait = FALSE)) {
+          while (!sem$wait(timeout_ms = 0)) {
 
             if (job$is_done) {
               output <- job$output
@@ -220,7 +284,7 @@ WebQueue <- R6Class(
         })
 
         if (!is.null(cnd)) worker$stop()
-        remove_semaphore(sem)
+        sem$remove()
         if (!is.null(cnd)) cnd_signal(cnd)
 
       } 
@@ -233,8 +297,8 @@ WebQueue <- R6Class(
 
         cnd <- catch_cnd({
 
-          # Start a Queue.
-          private$.jobqueue <- jobqueue::Queue$new(
+          # Start a `jobqueue`.
+          private$.jobqueue <- jobqueue::jobqueue(
             globals   = globals,
             packages  = packages,
             namespace = namespace,
@@ -250,7 +314,7 @@ WebQueue <- R6Class(
 
           later::run_now()
           if (!identical(private$.jobqueue$state, 'idle'))
-            stop('Unable to start jobqueue::Queue')  # nocov
+            stop('Unable to start a `jobqueue`')  # nocov
 
 
           # Start a Server.

README.md

@@ -38,7 +38,7 @@ pak::pak("cmmr/webqueue")
 ```r
 library(webqueue)
 
-wq <- WebQueue$new(~{ 'Hello world!\n' })
+wq <- webqueue(~{ 'Hello world!\n' })
 
 readLines('http://localhost:8080')
 #> [1] "Hello world!"
@@ -50,7 +50,7 @@ wq$stop()
 ## Query Parameters
 
 ```r
-wq <- WebQueue$new(~{ jsonlite::toJSON(.$ARGS) })
+wq <- webqueue(~{ jsonlite::toJSON(.$ARGS) })
 
 cat(RCurl::getURL('http://localhost:8080?myvar=123'))
 #> {"myvar":["123"]}
@@ -64,7 +64,7 @@ Accepts both GET and POST parameters.
 ## Simple API
 
 ```r
-wq <- WebQueue$new(
+wq <- webqueue(
   handler = function (req) {
     switch(
       EXPR = req$PATH_INFO,
@@ -106,7 +106,7 @@ See `vignette('interrupts')` for more detailed examples.
 ### Set a time limit
 
 ```r
-wq <- WebQueue$new(
+wq <- webqueue(
   handler = ~{ Sys.sleep(.$ARGS$s); 'Hello world!' }, 
   timeout = 1 )
 
@@ -124,13 +124,13 @@ wq$stop()
 ### Merge duplicate requests
 
 ```r
-wq <- WebQueue$new(
+wq <- webqueue(
   handler = function (req) { Sys.sleep(1); req$ARGS$x }, 
   copy_id = function (job) job$req$PATH_INFO )
 # ^^^^^^^   `copy_id` will be '/a' or '/b'
 
 # Fetch two URLs at the same time. '/b' path is merged.
-jq <- jobqueue::Queue$new(workers = 3L)$wait()   # vv
+jq <- jobqueue::jobqueue(workers = 3L)$wait()    # vv
 a1 <- jq$run({ RCurl::getURL('http://localhost:8080/a?x=first') })
 b1 <- jq$run({ RCurl::getURL('http://localhost:8080/b?x=second') })
 b2 <- jq$run({ RCurl::getURL('http://localhost:8080/b?x=third') })
@@ -145,13 +145,13 @@ wq$stop()
 
 ### Stop duplicate requests
 ```r
-wq <- WebQueue$new(
+wq <- webqueue(
   handler = function (req) { Sys.sleep(1); req$ARGS$x }, 
   stop_id = function (job) job$req$PATH_INFO )
 # ^^^^^^^   `stop_id` will be '/a' or '/b'
 
 # Fetch three URLs at the same time. '/b' path is stopped.
-jq <- jobqueue::Queue$new(workers = 3L)$wait()   # vv
+jq <- jobqueue::jobqueue(workers = 3L)$wait()    # vv
 a1 <- jq$run({ RCurl::getURL('http://localhost:8080/a?x=first') })
 b1 <- jq$run({ RCurl::getURL('http://localhost:8080/b?x=second') })
 b2 <- jq$run({ RCurl::getURL('http://localhost:8080/b?x=third') })