Add verbosity argument to req_peform_connection() (#632)

And warn when the buffer is discarded.

Fixes #599

Author: hadley

NEWS.md

@@ -45,10 +45,15 @@
 * `req_retry()` defaults to `max_tries = 2` when nethier `max_tries` nor
   `max_seconds` is set. If you want to disable retries, set `max_tries = 1`.
 
+* `req_perform_connection()` gains a `verbosity` argument, which is useful for
+  understanding exactly how data is streamed back to you (#599).
+
 * `req_url_query()` can control how spaces are encoded with `.space` (#432).
 
 * `resp_link_url()` handles multiple `Link` headers (#587).
 
+* `resp_stream_sse()` will warn if it recieves a partial event.
+
 * `url_parse()` parses relative URLs with new `base_url` argument (#449) and
   the uses faster and more correct `curl::curl_parse_url()` (#577).
 

R/req-perform-connection.R

@@ -1,4 +1,3 @@
-
 #' Perform a request and return a streaming connection
 #'
 #' @description
@@ -14,7 +13,7 @@
 #' than providing callbacks that the data is pushed to. This is useful if you
 #' want to do other work in between handling inputs from the stream.
 #'
-#' @inheritParams req_perform_stream
+#' @inheritParams req_perform
 #' @param blocking When retrieving data, should the connection block and wait
 #'   for the desired information or immediately return what it has (possibly
 #'   nothing)?
@@ -31,10 +30,12 @@
 #'
 #' # Always close the response when you're done
 #' close(resp)
-req_perform_connection <- function(req, blocking = TRUE) {
+req_perform_connection <- function(req, blocking = TRUE, verbosity = NULL) {
   check_request(req)
   check_bool(blocking)
+  # verbosity checked in req_verbosity_connection
 
+  req <- req_verbosity_connection(req, verbosity %||% httr2_verbosity())
   req <- auth_sign(req)
   req_prep <- req_prepare(req)
   handle <- req_handle(req_prep)
@@ -78,6 +79,30 @@ req_perform_connection <- function(req, blocking = TRUE) {
   resp
 }
 
+# Like req_verbosity() but we want to print the streaming body when it's
+# requested not when curl actually receives it
+req_verbosity_connection <- function(req, verbosity, error_call = caller_env()) {
+  if (!is_integerish(verbosity, n = 1) || verbosity < 0 || verbosity > 3) {
+    cli::cli_abort("{.arg verbosity} must 0, 1, 2, or 3.", call = error_call)
+  }
+
+  req <- switch(verbosity + 1,
+    req,
+    req_verbose(req),
+    req_verbose(req, body_req = TRUE),
+    req_verbose(req, body_req = TRUE, info = TRUE)
+  )
+  if (verbosity > 1) {
+    req <- req_policies(
+      req,
+      show_streaming_body = verbosity >= 2,
+      show_streaming_buffer = verbosity >= 3
+    )
+  }
+  req
+}
+
+
 req_perform_connection1 <- function(req, handle, blocking = TRUE) {
   stream <- curl::curl(req$url, handle = handle)
 

R/resp-stream-aws.R

@@ -9,11 +9,21 @@ resp_stream_aws <- function(resp, max_size = Inf) {
     include_trailer = FALSE
   )
 
-  if (!is.null(event_bytes)) {
-    parse_aws_event(event_bytes)
-  } else {
-    return(NULL)
+  if (is.null(event_bytes)) {
+    return()
+  }
+
+  event <- parse_aws_event(event_bytes)
+  if (resp_stream_show_body(resp)) {
+    # Emit header
+    for (key in names(event$headers)) {
+      log_stream(cli::style_bold(key), ": ", event$headers[[key]])
+    }
+    # Emit body
+    log_stream(jsonlite::toJSON(event$body, auto_unbox = TRUE, pretty = TRUE))
+    cli::cat_line()
   }
+  event
 }
 
 find_aws_event_boundary <- function(buffer) {
@@ -57,15 +67,15 @@ parse_aws_event <- function(bytes) {
 
   # headers
   headers <- list()
-  while(i <= 12 + header_length) {
+  while (i <= 12 + header_length) {
     name_length <- as.integer(read_bytes(1))
     name <- rawToChar(read_bytes(name_length))
     type <- as.integer(read_bytes(1))
 
     delayedAssign("length", parse_int(read_bytes(2)))
     value <- switch(type_enum(type),
-      'TRUE' = TRUE,
-      'FALSE' = FALSE,
+      "TRUE" = TRUE,
+      "FALSE" = FALSE,
       BYTE = parse_int(read_bytes(1)),
       SHORT = parse_int(read_bytes(2)),
       INTEGER = parse_int(read_bytes(4)),
@@ -95,7 +105,7 @@ parse_aws_event <- function(bytes) {
 # Helpers ----------------------------------------------------------------
 
 parse_int <- function(x) {
-  sum(as.integer(x) * 256 ^ rev(seq_along(x) - 1))
+  sum(as.integer(x) * 256^rev(seq_along(x) - 1))
 }
 
 parse_int64 <- function(x) {

R/resp-stream.R

@@ -26,11 +26,33 @@
 #' @param resp,con A streaming [response] created by [req_perform_connection()].
 #' @param kb How many kilobytes (1024 bytes) of data to read.
 #' @order 1
+#' @examples
+#' req <- request(example_url()) |>
+#'   req_template("GET /stream/:n", n = 5)
+#'
+#' con <- req |> req_perform_connection()
+#' while (!resp_stream_is_complete(con)) {
+#'   lines <- con |> resp_stream_lines(2)
+#'   cat(length(lines), " lines received\n", sep = "")
+#' }
+#' close(con)
+#'
+#' # You can also see what's happening by setting verbosity
+#' con <- req |> req_perform_connection(verbosity = 2)
+#' while (!resp_stream_is_complete(con)) {
+#'   lines <- con |> resp_stream_lines(2)
+#' }
+#' close(con)
 resp_stream_raw <- function(resp, kb = 32) {
   check_streaming_response(resp)
   conn <- resp$body
 
-  readBin(conn, raw(), kb * 1024)
+  out <- readBin(conn, raw(), kb * 1024)
+  if (resp_stream_show_body(resp)) {
+    log_stream("Streamed ", length(out), " bytes")
+    cli::cat_line()
+  }
+  out
 }
 
 #' @export
@@ -57,12 +79,18 @@ resp_stream_lines <- function(resp, lines = 1, max_size = Inf, warn = TRUE) {
     line <- resp_stream_oneline(resp, max_size, warn, encoding)
     if (length(line) == 0) {
       # No more data, either because EOF or req_perform_connection(blocking=FALSE).
-      # Either way, return what we have
-      return(lines_read)
+      # Either way we're done
+      break
     }
     lines_read <- c(lines_read, line)
     lines <- lines - 1
   }
+
+  if (resp_stream_show_body(resp)) {
+    log_stream(lines_read)
+    cli::cat_line()
+  }
+
   lines_read
 }
 
@@ -74,19 +102,25 @@ resp_stream_lines <- function(resp, lines = 1, max_size = Inf, warn = TRUE) {
 #' @order 1
 resp_stream_sse <- function(resp, max_size = Inf) {
   event_bytes <- resp_boundary_pushback(resp, max_size, find_event_boundary, include_trailer = FALSE)
-  if (!is.null(event_bytes)) {
-    parse_event(event_bytes)
-  } else {
-    return(NULL)
+  if (is.null(event_bytes)) {
+    return()
+  }
+
+  event <- parse_event(event_bytes)
+  if (resp_stream_show_body(resp)) {
+    for (key in names(event)) {
+      log_stream(cli::style_bold(key), ": ", pretty_json(event[[key]]))
+    }
+    cli::cat_line()
   }
+  event
 }
 
 #' @export
 #' @rdname resp_stream_raw
 resp_stream_is_complete <- function(resp) {
   check_response(resp)
-
-  !isIncomplete(resp$body)
+  length(resp$cache$push_back) == 0 && !isIncomplete(resp$body)
 }
 
 #' @export
@@ -187,16 +221,16 @@ find_event_boundary <- function(buffer) {
 
   boundary_end <- which(
     (left1 == 0x0A & buffer == 0x0A) | # \n\n
-    (left1 == 0x0D & buffer == 0x0D) | # \r\r
-    (left3 == 0x0D & left2 == 0x0A & left1 == 0x0D & buffer == 0x0A) # \r\n\r\n
+      (left1 == 0x0D & buffer == 0x0D) | # \r\r
+      (left3 == 0x0D & left2 == 0x0A & left1 == 0x0D & buffer == 0x0A) # \r\n\r\n
   )
 
   if (length(boundary_end) == 0) {
-    return(NULL)  # No event boundary found
+    return(NULL) # No event boundary found
   }
 
-  boundary_end <- boundary_end[1]  # Take the first occurrence
-  split_at <- boundary_end + 1  # Split at one after the boundary
+  boundary_end <- boundary_end[1] # Take the first occurrence
+  split_at <- boundary_end + 1 # Split at one after the boundary
   split_at
 }
 
@@ -217,7 +251,7 @@ split_buffer <- function(buffer, split_at) {
 #   the vector
 # @param include_trailer If TRUE, at the end of the response, if there are
 #   bytes after the last boundary, then return those bytes; if FALSE, then those
-#   bytes are silently discarded.
+#   bytes are discarded with a warning.
 resp_boundary_pushback <- function(resp, max_size, boundary_func, include_trailer) {
   check_streaming_response(resp)
   check_number_whole(max_size, min = 1, allow_infinite = TRUE)
@@ -228,8 +262,12 @@ resp_boundary_pushback <- function(resp, max_size, boundary_func, include_traile
   buffer <- resp$cache$push_back %||% raw()
   resp$cache$push_back <- raw()
 
-  print_buffer <- function(buf, label) {
-    # cat(label, ":", paste(sprintf("%02X", as.integer(buf)), collapse = " "), "\n", file = stderr())
+  if (resp_stream_show_buffer(resp)) {
+    print_buffer <- function(buf, label) {
+      cli::cat_line(" * ", label, ": ", paste(as.character(buf), collapse = " "))
+    }
+  } else {
+    print_buffer <- function(buf, label) {}
   }
 
   # Read chunks until we find an event or reach the end of input
@@ -261,24 +299,27 @@ resp_boundary_pushback <- function(resp, max_size, boundary_func, include_traile
       # one extra byte so we know to error.
       n = min(chunk_size, max_size - length(buffer) + 1)
     )
-
     print_buffer(chunk, "Received chunk")
 
-    # If we've reached the end of input, store the buffer and return NULL
     if (length(chunk) == 0) {
       if (!isIncomplete(resp$body)) {
         # We've truly reached the end of the connection; no more data is coming
-        if (include_trailer && length(buffer) > 0) {
-          return(buffer)
-        } else {
+        if (length(buffer) == 0) {
           return(NULL)
+        } else {
+          if (include_trailer) {
+            return(buffer)
+          } else {
+            cli::cli_warn("Premature end of input; ignoring final partial chunk")
+            return(NULL)
+          }
         }
+      } else {
+        # More data might come later; store the buffer and return NULL
+        print_buffer(buffer, "Storing incomplete buffer")
+        resp$cache$push_back <- buffer
+        return(NULL)
       }
-
-      # More data might come later
-      print_buffer(buffer, "Storing incomplete buffer")
-      resp$cache$push_back <- buffer
-      return(NULL)
     }
 
     # More data was received; combine it with existing buffer and continue the
@@ -322,7 +363,6 @@ parse_event <- function(event_data) {
 check_streaming_response <- function(resp,
                                      arg = caller_arg(resp),
                                      call = caller_env()) {
-
   check_response(resp, arg = arg, call = call)
 
   if (resp_body_type(resp) != "stream") {
@@ -353,3 +393,10 @@ isValid <- function(con) {
     error = function(cnd) FALSE
   )
 }
+
+resp_stream_show_body <- function(resp) {
+  resp$request$policies$show_streaming_body %||% FALSE
+}
+resp_stream_show_buffer <- function(resp) {
+  resp$request$policies$show_streaming_buffer %||% FALSE
+}

R/utils.R

@@ -34,7 +34,7 @@ modify_list <- function(.x, ..., error_call = caller_env()) {
     )
   }
 
-  
+
 
   out <- .x[!names(.x) %in% names(dots)]
   out <- c(out, compact(dots))
@@ -332,3 +332,20 @@ slice <- function(vector, start = 1, end = length(vector) + 1) {
 is_named_list <- function(x) {
   is_list(x) && (is_named(x) || length(x) == 0)
 }
+
+pretty_json <- function(x) {
+  parsed <- tryCatch(
+    jsonlite::parse_json(x),
+    error = function(e) NULL
+  )
+  if (is.null(parsed)) {
+    x
+  } else {
+    jsonlite::toJSON(parsed, auto_unbox = TRUE, pretty = TRUE)
+  }
+}
+
+log_stream <- function(..., prefix = "<< ") {
+  out <- gsub("\n", paste0("\n", prefix), paste0(prefix, ..., collapse = ""))
+  cli::cat_line(out)
+}