Add verbosity argument to req_peform_connection()

NEWS.md

@@ -1,5 +1,7 @@
 # httr2 (development version)
 
+* `resp_stream_sse()` will warn if it recieves a partial event.
+* `req_perform_connection()` gains a `verbosity` argument, which is useful for understanding exactly how data is streamed back to you (#599).
 * `req_perform_stream()` is superseded and likely on track for deprecation; `req_perform_connection()` is no longer experimental (#625)
 * `req_retry()` now optionally implements "circuit breaking" so that if requests to the same server fail many times (i.e. because the server is down), you can choose to immediately error rather than waiting (#370).
 * Export `is_online()` as thin wrapper around `curl::has_internet()` (#512).

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,26 @@
   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 = TRUE)
+  }
+  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 @@
     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_is_verbose(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 @@
 
   # 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 @@
 # 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

@@ -28,11 +28,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_is_verbose(resp)) {
+    log_stream("Streamed ", length(out), " bytes")
+    cli::cat_line()
+  }
+  out
 }
 
 #' @export
@@ -59,12 +81,18 @@
     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_is_verbose(resp)) {
+    log_stream(lines_read)
+    cli::cat_line()
+  }
+
   lines_read
 }
 
@@ -76,19 +104,25 @@
 #' @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_is_verbose(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
@@ -189,16 +223,16 @@
 
   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
 }
 
@@ -219,7 +253,7 @@
 #   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)
@@ -263,24 +297,27 @@
       # 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
@@ -324,7 +361,6 @@
 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") {
@@ -355,3 +391,7 @@
     error = function(cnd) FALSE
   )
 }
+
+resp_stream_is_verbose <- function(resp) {
+  resp$request$policies$show_streaming_body %||% FALSE
+}

R/utils.R

@@ -34,7 +34,7 @@
     )
   }
 
-  
+
 
   out <- .x[!names(.x) %in% names(dots)]
   out <- c(out, compact(dots))
@@ -332,3 +332,20 @@
 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)
+}