req_perform_stream(round = c("byte", "line"))

R/req-perform-stream.R

@@ -53,19 +53,30 @@ req_perform_stream <- function(req,
     }
 
     if (length(buf) > 0) {
+      # there are leftover bytes, but the stream is complete
+      # break the loop so that the callback() is given the
+      # whole buffer
+      if (!incomplete) {
+        break
+      }
+
       cut <- cut_points(buf)
       n <- length(cut)
       if (n) {
         continue <- isTRUE(callback(head(buf, n = cut[n])))
         buf <- tail(buf, n = -cut[n])
-      } else {
-        continue <- incomplete
       }
     } else {
       continue <- incomplete
     }
   }
 
+  # if there are leftover bytes and none of the callback()
+  # returned FALSE.
+  if (continue && length(buf)) {
+    callback(buf)
+  }
+
   data <- curl::handle_data(handle)
   new_response(
     method = req_method_get(req),

tests/testthat/test-req-perform-stream.R

@@ -34,7 +34,7 @@ test_that("can supply custom rounding", {
       buffer_kb = 0.1,
       round = function(bytes) if (length(bytes) > 100) 100 else integer()
     )
-  expect_equal(lengths(out), rep(100, 10))
+  expect_equal(lengths(out), c(rep(100, 10), 24))
 })
 
 test_that("eventually terminates even if never rounded", {
@@ -50,7 +50,7 @@ test_that("eventually terminates even if never rounded", {
       buffer_kb = 0.1,
       round = function(bytes) integer()
     )
-  expect_equal(length(out), 0)
+  expect_equal(length(out), 1024)
 })
 
 test_that("as_round_function checks its inputs", {