Improve synchronization in streaming tests

Author: shikokuchuo

tests/testthat/helper-sync.R

@@ -1,6 +1,7 @@
 sync_req <- function(name, .env = parent.frame()) {
   skip_on_cran()
   skip_if_not_installed("nanonext")
+  skip_if_not_installed("later")
 
   if (missing(name) || !is.character(name)) {
     cli::cli_abort(
@@ -14,19 +15,13 @@ sync_req <- function(name, .env = parent.frame()) {
   nanonext::pipe_notify(sock, cv, add = TRUE)
   nanonext::listen(sock, url = sprintf("ipc:///tmp/nanonext%s", name))
 
-  function(
-    expr = {},
-    timeout = 1000L
-  ) {
+  function(resp, timeout = 1000L) {
     if (!connected) {
       nanonext::until(cv, timeout)
       connected <<- TRUE
     }
-    ctx <- nanonext::context(sock)
-    saio <- nanonext::send_aio(ctx, 0L, mode = 2L)
-    expr
-    nanonext::call_aio(nanonext::recv_aio(ctx, mode = 8L, timeout = timeout))
-    nanonext::msleep(50L) # wait, as nanonext messages can return faster than side effects (e.g. stream)
+    nanonext::send(sock, 0L, mode = 2L, block = timeout)
+    wait_for_http_data(resp, timeout / 1000)
   }
 }
 
@@ -44,17 +39,56 @@ sync_rep <- function(name, .env = parent.frame()) {
   nanonext::pipe_notify(sock, cv, add = TRUE)
   nanonext::dial(sock, url = sprintf("ipc:///tmp/nanonext%s", name))
 
-  function(
-    expr = {},
-    timeout = 1000L
-  ) {
+  function(expr = {}, timeout = 1000L) {
     if (!connected) {
       nanonext::until(cv, timeout)
       connected <<- TRUE
     }
-    ctx <- nanonext::context(sock)
-    nanonext::call_aio(nanonext::recv_aio(ctx, mode = 8L, timeout = timeout))
+    nanonext::recv(sock, mode = 8L, block = timeout)
     expr
-    nanonext::send(ctx, 0L, mode = 2L, block = TRUE)
   }
 }
+
+wait_for_http_data <- function(resp, timeout_s) {
+  if (resp$body$is_complete()) {
+    return(invisible(TRUE))
+  }
+
+  deadline <- as.double(Sys.time()) + timeout_s
+
+  while ((remaining <- deadline - as.double(Sys.time())) > 0) {
+    fdset <- resp$body$get_fdset()
+    if (length(fdset$reads) == 0) {
+      return(invisible(FALSE))
+    }
+
+    fd_ready <- FALSE
+    later::later_fd(
+      func = function(ready) {
+        fd_ready <<- any(ready, na.rm = TRUE)
+      },
+      readfds = fdset$reads,
+      timeout = remaining
+    )
+    later::run_now(remaining)
+
+    if (!fd_ready) {
+      break
+    } # Timeout
+
+    # Try to actually read data from FD
+    chunk <- resp$body$read(256)
+
+    if (length(chunk) > 0) {
+      # Append new data to push_back so tests can read it
+      resp$cache$push_back <- c(resp$cache$push_back, chunk)
+      return(invisible(TRUE))
+    }
+
+    if (resp$body$is_complete()) {
+      return(invisible(TRUE))
+    }
+  }
+
+  invisible(FALSE)
+}

tests/testthat/test-resp-stream.R

@@ -83,7 +83,7 @@ test_that("can join lines across multiple reads", {
   out <- resp_stream_lines(resp1)
   expect_equal(out, character())
 
-  sync()
+  sync(resp1)
   out <- resp_stream_lines(resp1)
   expect_equal(out, "This is a complete sentence.")
 })
@@ -121,7 +121,7 @@ test_that("handles line endings of multiple kinds", {
 
   for (expected in expected_values) {
     rlang::inject(expect_equal(resp_stream_lines(resp1), !!expected))
-    sync()
+    sync(resp1)
   }
   expect_warning(out <- resp_stream_lines(resp1), "incomplete final line")
   expect_equal(out, "eof without line ending")
@@ -233,11 +233,11 @@ test_that("can join sse events across multiple reads", {
   expect_equal(out, NULL)
   expect_equal(resp1$cache$push_back, charToRaw("data: 1\n"))
 
-  sync()
+  sync(resp1)
   out <- resp_stream_sse(resp1)
   expect_equal(out, NULL)
 
-  sync()
+  sync(resp1)
   out <- resp_stream_sse(resp1)
   expect_equal(out, list(type = "message", data = "1\n2", id = ""))
   expect_equal(resp1$cache$push_back, charToRaw("data: 3\n\n"))