Implements event-driven async for req_perform_promise() (#580)

Fixes #579.

DESCRIPTION

@@ -37,7 +37,7 @@ Suggests:
     jose,
     jsonlite,
     knitr,
-    later,
+    later (>= 1.3.2.9001),
     paws.common,
     promises,
     rmarkdown,
@@ -54,3 +54,4 @@ Config/testthat/start-first: resp-stream, req-perform
 Encoding: UTF-8
 Roxygen: list(markdown = TRUE)
 RoxygenNote: 7.3.2
+Remotes: r-lib/later

NEWS.md

@@ -1,5 +1,7 @@
 # httr2 (development version)
 
+* `req_perform_promise()` upgraded to use event-driven async based on waiting efficiently on curl socket activity (#579).
+
 # httr2 1.0.6
 
 * Fix stochastic test failure, particularly on CRAN (#572)

R/req-promise.R

@@ -9,6 +9,10 @@
 #' [promises package documentation](https://rstudio.github.io/promises/articles/promises_01_motivation.html)
 #' for more details on how to work with the resulting promise object.
 #'
+#' If using together with [later::with_temp_loop()] or other private event loops,
+#' a new curl pool made by [curl::new_pool()] should be created for requests made
+#' within the loop to ensure that only these requests are being polled by the loop.
+#'
 #' Like with [req_perform_parallel()], exercise caution when using this function;
 #' it's easy to pummel a server with many simultaneous requests. Also, not all servers
 #' can handle more than 1 request at a time, so the responses may still return
@@ -69,6 +73,15 @@ req_perform_promise <- function(req,
   check_installed(c("promises", "later"))
   check_string(path, allow_null = TRUE)
 
+  if (missing(pool)) {
+    if (!identical(later::current_loop(), later::global_loop())) {
+      cli::cli_abort(c(
+        "Must supply {.arg pool} when calling {.code later::with_temp_loop()}.",
+        i = "Do you need {.code pool = curl::new_pool()}?"
+      ))
+    }
+  }
+
   promises::promise(
     function(resolve, reject) {
       perf <- PerformancePromise$new(
@@ -131,12 +144,19 @@ ensure_pool_poller <- function(pool, reject) {
   monitor <- pool_poller_monitor(pool)
   if (monitor$already_going()) return()
 
-  poll_pool <- function() {
+  poll_pool <- function(ready) {
     tryCatch(
       {
         status <- curl::multi_run(0, pool = pool)
         if (status$pending > 0) {
-          later::later(poll_pool, delay = 0.1, loop = later::global_loop())
+          fds <- curl::multi_fdset(pool = pool)
+          later::later_fd(
+            func = poll_pool,
+            readfds = fds$reads,
+            writefds = fds$writes,
+            exceptfds = fds$exceptions,
+            timeout = fds$timeout
+          )
         } else {
           monitor$ending()
         }

tests/testthat/_snaps/req-promise.md

@@ -0,0 +1,9 @@
+# req_perform_promise uses the default loop
+
+    Code
+      p4 <- req_perform_promise(request_test("/get"))
+    Condition
+      Error in `req_perform_promise()`:
+      ! Must supply `pool` when calling `later::with_temp_loop()`.
+      i Do you need `pool = curl::new_pool()`?
+

tests/testthat/test-req-promise.R

@@ -110,3 +110,39 @@ test_that("req_perform_promise can use non-default pool", {
   p2_value <- extract_promise(p2)
   expect_equal(resp_status(p2_value), 200)
 })
+
+test_that("req_perform_promise uses the default loop", {
+  # The main reason for temp loops is to allow an asynchronous operation to be
+  # created, waited on, and resolved/rejected inside of a synchronous function,
+  # all without affecting any asynchronous operations that existed before the
+  # temp loop was created.
+
+  # This can't proceed within the temp loop
+  p1 <- req_perform_promise(request_test("/get"))
+
+  later::with_temp_loop({
+    # You can create an async response with explicit pool=NULL, but it can't
+    # proceed until the temp loop is over
+    p2 <- req_perform_promise(request_test("/get"), pool = NULL)
+
+    # You can create an async response with explicit pool=pool, and it can
+    # proceed as long as that pool was first used inside of the temp loop
+    p3 <- req_perform_promise(request_test("/get"), pool = curl::new_pool())
+
+    # You can't create an async response in the temp loop without explicitly
+    # specifying a pool
+    expect_snapshot(p4 <- req_perform_promise(request_test("/get")), error = TRUE)
+
+    # Like I said, you can create this, but it won't work until we get back
+    # outside the temp loop
+    expect_null(extract_promise(p2, timeout = 1))
+
+    # This works fine inside the temp loop, because its pool was first used
+    # inside
+    expect_equal(resp_status(extract_promise(p3, timeout = 1)), 200)
+  })
+
+  # These work fine now that we're back outside the temp loop
+  expect_equal(resp_status(extract_promise(p1, timeout = 1)), 200)
+  expect_equal(resp_status(extract_promise(p2, timeout = 1)), 200)
+})