Result and Error Handling
In jobqueue: Run Interruptible Code Asynchronously

Reacting to Results

library(jobqueue)
jq <- jobqueue()

my_job <- jq$run({ 42 })

# ===  A 'jobqueue'-style callback  ============
my_job$on('done', ~message('Result = ', .$result))
#> Result = 42

# ===  A 'promises'-style callback  ============
my_job %...>% message('Result = ', .)
#> Result = 42

Output vs Result

When expr is finished evaluating, the result is assigned to <job>$output.

job <- jq$run({ 42 })

job$output
#> [1] 42

By default, <job>$result will return the exact same value as <job>$output.

job$result
#> [1] 42

However, while <job>$output is a fixed value, <job>$result is highly configurable. Most notably by the reformat and signal parameters.

Reformatting Output

Use reformat = function (job) {...} to define what should be returned by <job>$result.

Important

You may access job$output in your reformat function.
DO NOT access job$result, which would start an infinite recusion.

reformat can be set in several places:

jobqueue(reformat = http_wrap)
<jobqueue>$run(reformat = http_wrap)
<job>$reformat <- http_wrap

Transformation

http_wrap <- function (job) {
  result <- job$output
  paste0('{"status":200,"body":{"result":', result, '}}')
}

job <- jq$run({ 42 }, reformat = http_wrap)
cat(job$result)
#> {"status":200,"body":{"result":42}}

`job` Details

with_id <- function (job) {
  list(result = job$output, id = job$id)
}

job <- jq$run({ 42 }, reformat = with_id, id = 'abc')
dput(job$result)
#> list(result = 42, id = "abc")

Non-blocking

<job>$output blocks until the job is done. If you want <job>$result to not block, you can return a placeholder value instead.

reformat <- function (job) {
  if (job$is_done) { job$output                 }
  else             { 'result not available yet' }
}

job <- jq$run({ Sys.sleep(5); 42 }, reformat = reformat)
job$result
#> [1] "result not available yet"

job$state
#> [1] "running"

Signaling Errors

By default, errors and interrupts will be caught by the jobqueue and returned as a condition object from <job>$result.

expr <- quote(stop('error XYZ'))

job <- jq$run(expr)
x   <- job$result

inherits(x, 'condition')
#> TRUE

x
#> <simpleError in eval(expr, envir, enclos): error XYZ>

If you want those errors to continue propagating, set signal = TRUE.

job <- jq$run(expr, signal = TRUE)
x   <- job$result
#> Error:
#> ! stop("error XYZ")
#> Caused by error:
#> ! error XYZ
#> Run `rlang::last_trace()` to see where the error occurred.

These signals can be caught as usual by try() and tryCatch().

print_cnd <- function (cnd) cli::cli_text('Caught {.type {cnd}}.')

job <- jq$run({ Sys.sleep(5) }, signal = TRUE)$stop()
res <- tryCatch(job$result, error = print_cnd, interrupt = print_cnd)
#> Caught an <interrupt> object.

res
#> NULL

job <- jq$run({ stop('error XYZ') }, signal = TRUE)
res <- tryCatch(job$result, error = print_cnd, interrupt = print_cnd)
#> Caught an <error> object.

res
#> NULL

You can also signal some conditions while catching others. Perhaps stopping a job is part of the "normal" workflow and shouldn't be lumped in with other errors.

job <- jq$run({ Sys.sleep(5) }, signal = 'error')$stop()
res <- tryCatch(job$result, error = print_cnd)

res
#> <interrupt: job stopped by user>

job <- jq$run({ stop('error XYZ') }, signal = 'error')
res <- tryCatch(job$result, error = print_cnd)
#> Caught an <error> object.

res
#> NULL

Promise Handling

You can pass a job directly to any function that expects a promise object, such as promises::then or promises::%...>%. These functions are re-exported by jobqueue, so calling library(promises) is optional.

jq$run({ 42 }) %...>% message
#> 42

By default, errors and interrupts are caught by jobqueue and used for promise fulfillment.

onFulfilled <- function (value) cli::cli_text('Fulfilled with {.type {value}}.')
onRejected  <- function (err)   cli::cli_text('Rejected with {.type {err}}.')

job <- jq$run({ Sys.sleep(5) })$stop()
job %>% then(onFulfilled, onRejected)
#> Fulfilled with an <interrupt> object.

job <- jq$run({ stop('error XYZ') })
job %>% then(onFulfilled, onRejected)
#> Fulfilled with an <error> object.

Setting signal = TRUE will cause errors and interrupts be sent to the promise's rejection handler instead.

job <- jq$run({ Sys.sleep(5) }, signal = TRUE)$stop()
job %>% then(onFulfilled, onRejected)
#> Rejected with an <interrupt> object.

job <- jq$run({ stop('error XYZ') }, signal = TRUE)
job %>% then(onFulfilled, onRejected)
#> Rejected with an <error> object.