diff --git a/DESCRIPTION b/DESCRIPTION index 1bbc5daf9..9111641f6 100644 --- a/DESCRIPTION +++ b/DESCRIPTION @@ -32,6 +32,7 @@ Suggests: bench, clipr, covr, + digest, docopt, httpuv, jose, diff --git a/NEWS.md b/NEWS.md index 481cc96ab..aa6b30c2a 100644 --- a/NEWS.md +++ b/NEWS.md @@ -15,6 +15,10 @@ e.g., `application/problem+json` (@cgiachalis, #782). * `req_body_form()` and `req_url_query()` no longer error with "C stack usage is too close to the limit" when given very long string values (#805). * `req_body_form()` now creates a valid empty request body when no parameters are provided (@arcresu, #836). +* `resp_stream_aws()` now parses `byte`, `short`, and `integer` headers as signed integers, matching the AWS event-stream specification (previously they were incorrectly read as unsigned). +* `resp_stream_lines()` no longer treats a bare carriage return (CR) as a line ending; only LF and CRLF terminate lines, which is what every modern streaming source produces. +* `resp_stream_lines()` no longer warns when the stream ends without a final line terminator (which is routine when streaming), and its `warn` argument is (softly) deprecated. +* `resp_stream_lines()`, `resp_stream_sse()`, and `resp_stream_aws()` now decode whole chunks at a time and hold the results in a queue, instead of rescanning and recopying the buffer for every line or event. This makes memory use and run time scale linearly rather than quadratically with the response size, so large streams use dramatically less memory and run much faster (e.g. reading a 1 MB response of short lines is now around 200x faster and allocates around 180x less memory) (#704). * `req_throttle()` can now enforce multiple rate limits at once: supply a vector to `capacity` (and `fill_time_s`) to create one token bucket per limit, and each request must satisfy all of them (#555). * `req_auth_aws_v4()` now correctly signs URLs containing encoded slashes (`%2F`) in path segments, such as ARNs in AWS Bedrock API paths (@thisisnic, #842). * `req_error()` is now applied to responses retrieved from the cache, so a custom `is_error` callback is respected on cache hits (#806). diff --git a/R/req-perform-connection.R b/R/req-perform-connection.R index 511a09b37..8525c241b 100644 --- a/R/req-perform-connection.R +++ b/R/req-perform-connection.R @@ -252,3 +252,17 @@ StreamingBody <- R6::R6Class( conn = NULL ) ) + +# isOpen doesn't work for two reasons: +# 1. It errors if con has been closed, rather than returning FALSE +# 2. If returns TRUE if con has been closed and a new connection opened +# +# So instead we retrieve the connection from its number and compare to the +# original connection. This works because connections have an undocumented +# external pointer. +isValid <- function(con) { + tryCatch( + identical(getConnection(con), con), + error = function(cnd) FALSE + ) +} diff --git a/R/resp-stream-aws.R b/R/resp-stream-aws.R index 2163e65af..91142f5ad 100644 --- a/R/resp-stream-aws.R +++ b/R/resp-stream-aws.R @@ -1,17 +1,15 @@ #' @export #' @rdname resp_stream_raw -#' @order 2 +#' @order 4 resp_stream_aws <- function(resp, max_size = Inf) { - event_bytes <- resp_boundary_pushback( - resp = resp, - max_size = max_size, - boundary_func = find_aws_event_boundary, - include_trailer = FALSE - ) + splitter <- init_streaming_response(resp, AwsSplitter) + check_number_whole(max_size, min = 1, allow_infinite = TRUE) - if (is.null(event_bytes)) { + blocks <- stream_pull(resp, 1, splitter, max_size) + if (length(blocks) == 0L) { return() } + event_bytes <- blocks[[1L]] event <- parse_aws_event(event_bytes) if (resp_stream_show_body(resp)) { @@ -26,24 +24,50 @@ resp_stream_aws <- function(resp, max_size = Inf) { event } -find_aws_event_boundary <- function(buffer) { - # No valid AWS event message is less than 16 bytes - if (length(buffer) < 16) { - return(NULL) - } - - # Read first 4 bytes as a big endian number - event_size <- parse_int(buffer[1:4]) - if (event_size > length(buffer)) { - return(NULL) +AwsSplitter <- R6::R6Class( + "AwsSplitter", + inherit = StreamSplitter, + public = list( + name = "resp_stream_aws()", + find_boundaries = function(buffer) find_aws_event_boundaries(buffer) + ) +) + +# Find every complete AWS event in a buffer by walking the 4-byte big-endian +# length prefix at the start of each event. Returns a vector of split points +# (the position one past the end of each complete event). +find_aws_event_boundaries <- function(buffer) { + n <- length(buffer) + splits <- double() + pos <- 1 + repeat { + # No valid AWS event message is less than 16 bytes. + if (n - pos + 1L < 16L) { + break + } + # Read the first 4 bytes of the event as a big endian number. + event_size <- parse_int(buffer[pos:(pos + 3L)]) + if (event_size > n - pos + 1L) { + break + } + pos <- pos + event_size + splits[[length(splits) + 1L]] <- pos } - - event_size + 1 + splits } -# Implementation from https://github.com/lifion/lifion-aws-event-stream/blob/develop/lib/index.js -# This is technically buggy because it takes the header_length as a lower bound -# but this shouldn't cause problems in practive +# Parse a single AWS event-stream message (content type +# application/vnd.amazon.eventstream). The binary format is documented by AWS: +# * https://smithy.io/2.0/aws/amazon-eventstream.html (canonical protocol spec) +# * https://docs.aws.amazon.com/lexv2/latest/dg/event-stream-encoding.html +# Reference implementation: https://github.com/awslabs/aws-eventstream-java +# +# Key details: all integers are big-endian; the prelude (total + header lengths) +# and the whole message each end in a GZIP/zlib CRC32; header value types +# byte/short/integer/long are signed; timestamp is an int64 of epoch millis. +# +# We treat header_length as a lower bound rather than an exact count; this is +# lenient but harmless and matches some reference implementations. parse_aws_event <- function(bytes) { i <- 1 read_bytes <- function(n) { @@ -80,9 +104,9 @@ parse_aws_event <- function(bytes) { type_enum(type), "TRUE" = TRUE, "FALSE" = FALSE, - BYTE = parse_int(read_bytes(1)), - SHORT = parse_int(read_bytes(2)), - INTEGER = parse_int(read_bytes(4)), + BYTE = parse_int(read_bytes(1), signed = TRUE), + SHORT = parse_int(read_bytes(2), signed = TRUE), + INTEGER = parse_int(read_bytes(4), signed = TRUE), LONG = parse_int64(read_bytes(8)), BYTE_ARRAY = read_bytes(length), CHARACTER = rawToChar(read_bytes(length)), @@ -108,8 +132,13 @@ parse_aws_event <- function(bytes) { # Helpers ---------------------------------------------------------------- -parse_int <- function(x) { - sum(as.integer(x) * 256^rev(seq_along(x) - 1)) +parse_int <- function(x, signed = FALSE) { + v <- sum(as.integer(x) * 256^rev(seq_along(x) - 1)) + if (signed && v >= 2^(8 * length(x) - 1)) { + # Interpret as two's complement. + v <- v - 2^(8 * length(x)) + } + v } parse_int64 <- function(x) { @@ -119,7 +148,7 @@ parse_int64 <- function(x) { } type_enum <- function(value) { - if (value < 0 || value > 10) { + if (value < 0 || value > 9) { cli::cli_abort("Unsupported type {value}.", .internal = TRUE) } @@ -138,13 +167,6 @@ type_enum <- function(value) { ) } -hex_to_raw <- function(x) { - x <- gsub("(\\s|\n)+", "", x) - - pairs <- substring(x, seq(1, nchar(x), by = 2), seq(2, nchar(x), by = 2)) - as.raw(strtoi(pairs, 16L)) -} - raw_to_hex <- function(x) { paste(as.character(x), collapse = "") } diff --git a/R/resp-stream-lines.R b/R/resp-stream-lines.R new file mode 100644 index 000000000..704a6869c --- /dev/null +++ b/R/resp-stream-lines.R @@ -0,0 +1,57 @@ +#' @export +#' @rdname resp_stream_raw +#' @param lines The maximum number of lines to return at once. +#' @param warn `r lifecycle::badge("deprecated")` `resp_stream_lines()` no longer +#' warns when the connection ends without a final EOL, so this argument is +#' ignored. +#' @order 2 +resp_stream_lines <- function( + resp, + lines = 1, + max_size = Inf, + warn = deprecated() +) { + splitter <- init_streaming_response(resp, LineSplitter) + check_number_whole(lines, min = 0, allow_infinite = TRUE) + check_number_whole(max_size, min = 1, allow_infinite = TRUE) + if (lifecycle::is_present(warn) && !isFALSE(warn)) { + lifecycle::deprecate_warn("1.2.3", "resp_stream_lines(warn)") + } + + if (lines == 0) { + return(character()) + } + + encoding <- env_cache(resp$cache, "stream_encoding", resp_encoding(resp)) + blocks <- stream_pull(resp, lines, splitter, max_size) + lines_read <- stream_parse_lines(blocks, encoding) + if (resp_stream_show_body(resp)) { + log_stream(lines_read) + } + lines_read +} + +# Splits a stream into lines terminated by LF (and hence CRLF) +LineSplitter <- R6::R6Class( + "LineSplitter", + inherit = StreamSplitter, + public = list( + name = "resp_stream_lines()", + find_boundaries = function(buffer) { + grepRaw(as.raw(0x0A), buffer, fixed = TRUE, all = TRUE) + 1L + }, + # At end of stream, a trailing line without a terminator is still a line. + finish = function(remainder) { + if (length(remainder) == 0L) list() else list(remainder) + } + ) +) + +# Decode raw line blocks (each a line plus its trailing LF or CRLF) into a +# character vector in `encoding`, dropping the terminators. +stream_parse_lines <- function(blocks, encoding) { + text <- vapply(blocks, rawToChar, character(1)) + Encoding(text) <- "bytes" + text <- iconv(text, encoding, "UTF-8") + sub("\r?\n$", "", text) +} diff --git a/R/resp-stream-sse.R b/R/resp-stream-sse.R new file mode 100644 index 000000000..24ce34f61 --- /dev/null +++ b/R/resp-stream-sse.R @@ -0,0 +1,172 @@ +#' @param max_size The maximum number of bytes to buffer while waiting for a +#' line or event boundary; if exceeded, an error is thrown. This limit is +#' approximate: to spot a boundary httr2 may buffer a handful of bytes beyond +#' `max_size` (e.g. the bytes of the delimiter itself). +#' @export +#' @rdname resp_stream_raw +#' @order 3 +resp_stream_sse <- function(resp, max_size = Inf) { + splitter <- init_streaming_response(resp, SseSplitter) + check_number_whole(max_size, min = 1, allow_infinite = TRUE) + + repeat { + blocks <- stream_pull(resp, 1, splitter, max_size) + if (length(blocks) == 0L) { + return() + } + event_bytes <- blocks[[1L]] + + if (resp_stream_show_buffer(resp)) { + log_stream( + cli::rule("Raw server sent event"), + "\n", + rawToChar(event_bytes), + prefix = "* " + ) + } + + event <- parse_event(event_bytes) + if (!is.null(event)) break + } + + 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 +} + +# Splits a server-sent event stream into events at their boundaries. +SseSplitter <- R6::R6Class( + "SseSplitter", + inherit = StreamSplitter, + public = list( + name = "resp_stream_sse()", + find_boundaries = function(buffer) find_event_boundaries(buffer) + ) +) + +# Find every event boundary in a buffer, returning a vector of split points +# (the position one past the end of each boundary). Events may be separated by +# a double LF, a double CR, or a double CRLF. +# +# Example: +# find_event_boundaries(charToRaw("data: 1\n\nid: 12345")) +# Returns: +# 9L (so the first event is bytes 1:8, "data: 1\n\n") +find_event_boundaries <- function(buffer) { + nn <- grepRaw("\n\n", buffer, fixed = TRUE, all = TRUE) + rr <- grepRaw("\r\r", buffer, fixed = TRUE, all = TRUE) + rnrn <- grepRaw("\r\n\r\n", buffer, fixed = TRUE, all = TRUE) + + # Fast paths for the common case of a single, consistent delimiter. + if (length(rr) == 0L && length(rnrn) == 0L) { + return(nn + 2L) + } + if (length(nn) == 0L && length(rnrn) == 0L) { + return(rr + 2L) + } + if (length(nn) == 0L && length(rr) == 0L) { + return(rnrn + 4L) + } + + # Mixed delimiters: merge candidates and walk them left to right, taking + # non-overlapping boundaries. + starts <- c(nn, rr, rnrn) + ends <- c(nn + 1L, rr + 1L, rnrn + 3L) + o <- order(starts) + starts <- starts[o] + ends <- ends[o] + + keep <- logical(length(starts)) + consumed <- 0L + for (k in seq_along(starts)) { + if (starts[k] > consumed) { + keep[k] <- TRUE + consumed <- ends[k] + } + } + ends[keep] + 1L +} + +# https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation +parse_event <- function(event_data) { + if (is.raw(event_data)) { + # Streams must be decoded using the UTF-8 decode algorithm. + str_data <- rawToChar(event_data) + Encoding(str_data) <- "UTF-8" + } else { + # for testing + str_data <- event_data + } + + # The stream must then be parsed by reading everything line by line, with a + # U+000D CARRIAGE RETURN U+000A LINE FEED (CRLF) character pair, a single + # U+000A LINE FEED (LF) character not preceded by a U+000D CARRIAGE RETURN + # (CR) character, and a single U+000D CARRIAGE RETURN (CR) character not + # followed by a U+000A LINE FEED (LF) character being the ways in + # which a line can end. + lines <- strsplit(str_data, "\r\n|\r|\n")[[1]] + + # When a stream is parsed, a data buffer, an event type buffer, and a + # last event ID buffer must be associated with it. They must be initialized + # to the empty string. + data <- "" + type <- "" + last_id <- "" + + # If the line starts with a U+003A COLON character (:) - Ignore the line. + lines <- lines[!grepl("^:", lines)] + + # If the line contains a U+003A COLON character (:) + # * Collect the characters on the line before the first U+003A COLON + # character (:), and let field be that string. + # * Collect the characters on the line after the first U+003A COLON character + # (:), and let value be that string. If value starts with a U+0020 SPACE + # character, remove it from value. + m <- regexec("([^:]*)(: ?)?(.*)", lines) + matches <- regmatches(lines, m) + keys <- c("event", vapply(matches, function(x) x[2], character(1))) + values <- c("message", vapply(matches, function(x) x[4], character(1))) + + for (i in seq_along(matches)) { + key <- matches[[i]][2] + value <- matches[[i]][4] + + if (key == "event") { + # Set the event type buffer to field value. + type <- value + } else if (key == "data") { + # Append the field value to the data buffer, then append a single + # U+000A LINE FEED (LF) character to the data buffer. + data <- paste0(data, value, "\n") + } else if (key == "id") { + # If the field value does not contain U+0000 NULL, then set the last + # event ID buffer to the field value. Otherwise, ignore the field. + last_id <- value + } + } + + # If the data buffer is an empty string, set the data buffer and the event + # type buffer to the empty string and return. + if (data == "") { + return() + } + + # If the data buffer's last character is a U+000A LINE FEED (LF) character, + # then remove the last character from the data buffer. + if (grepl("\n$", data)) { + data <- substr(data, 1, nchar(data) - 1) + } + if (type == "") { + type <- "message" + } + + list( + type = type, + data = data, + id = last_id + ) +} diff --git a/R/resp-stream.R b/R/resp-stream.R index 36d0b0b14..253d8a564 100644 --- a/R/resp-stream.R +++ b/R/resp-stream.R @@ -46,7 +46,8 @@ #' } #' close(con) resp_stream_raw <- function(resp, kb = 32) { - check_streaming_response(resp) + init_streaming_response(resp, RawSplitter) + check_number_decimal(kb, min = 0, allow_infinite = FALSE) out <- resp$body$read(kb * 1024) if (resp_stream_show_body(resp)) { @@ -58,93 +59,26 @@ resp_stream_raw <- function(resp, kb = 32) { #' @export #' @rdname resp_stream_raw -#' @param lines The maximum number of lines to return at once. -#' @param warn Like [readLines()]: warn if the connection ends without a final -#' EOL. -#' @order 1 -resp_stream_lines <- function(resp, lines = 1, max_size = Inf, warn = TRUE) { - check_streaming_response(resp) - check_number_whole(lines, min = 0, allow_infinite = TRUE) - check_number_whole(max_size, min = 1, allow_infinite = TRUE) - check_logical(warn) - - if (lines == 0) { - # If you want to do that, who am I to judge? - return(character()) - } - - encoding <- resp_encoding(resp) - - lines_read <- character(0) - while (lines > 0) { - 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 we're done - break - } - lines_read <- c(lines_read, line) - lines <- lines - 1 - } - - if (resp_stream_show_body(resp)) { - log_stream(lines_read) - } - - lines_read +#' @order 6 +resp_stream_is_complete <- function(resp) { + init_streaming_response(resp) + !stream_has_buffered(resp) && resp$body$is_complete() } - -#' @param max_size The maximum number of bytes to buffer; once this number of -#' bytes has been exceeded without a line/event boundary, an error is thrown. -#' @export -#' @rdname resp_stream_raw -#' @order 1 -resp_stream_sse <- function(resp, max_size = Inf) { - repeat { - event_bytes <- resp_boundary_pushback( - resp, - max_size, - find_event_boundary, - include_trailer = FALSE - ) - if (is.null(event_bytes)) { - return() - } - - if (resp_stream_show_buffer(resp)) { - log_stream( - cli::rule("Raw server sent event"), - "\n", - rawToChar(event_bytes), - prefix = "* " - ) - } - - event <- parse_event(event_bytes) - if (!is.null(event)) break - } - - if (resp_stream_show_body(resp)) { - for (key in names(event)) { - log_stream(cli::style_bold(key), ": ", pretty_json(event[[key]])) - } - cli::cat_line() +# Is there any data still buffered in the response that hasn't been returned to +# the user yet? (Either raw bytes or decoded-but-unserved blocks.) +stream_has_buffered <- function(resp) { + cache <- resp$cache + if (length(cache$stream_buffer) > 0) { + return(TRUE) } - event -} - -#' @export -#' @rdname resp_stream_raw -resp_stream_is_complete <- function(resp) { - check_response(resp) - length(resp$cache$push_back) == 0 && resp$body$is_complete() + cache$stream_pos <= length(cache$stream_queue) } #' @export #' @param ... Not used; included for compatibility with generic. #' @rdname resp_stream_raw -#' @order 3 +#' @order 5 close.httr2_response <- function(con, ...) { check_response(con) @@ -155,301 +89,181 @@ close.httr2_response <- function(con, ...) { invisible() } -resp_stream_oneline <- function(resp, max_size, warn, encoding) { - repeat { - line_bytes <- resp_boundary_pushback( - resp, - max_size, - find_line_boundary, - include_trailer = TRUE - ) - if (is.null(line_bytes)) { - return(character()) - } +# Streaming engine ------------------------------------------------------------ - eat_next_lf <- resp$cache$resp_stream_oneline_eat_next_lf - resp$cache$resp_stream_oneline_eat_next_lf <- FALSE +# How many bytes to read from the connection at a time when streaming. +stream_chunk_bytes <- 64L * 1024L - if (identical(line_bytes, as.raw(0x0A)) && isTRUE(eat_next_lf)) { - # We hit that special edge case, see below - next +# Reads from a streaming response and uses `splitter` to divide the byte stream +# into blocks, returning up to `n` of them in a list. The byte-level details - +# where blocks begin and end, how trailing bytes are handled at the end of the +# stream, and how many bytes to read at a time - all live in the `splitter` +# (a `StreamSplitter`), so that this loop can be shared by lines, server-sent +# events, and AWS events. +# +# A whole chunk is split at once and the resulting blocks are held in a queue +# served by an index pointer, so that reading a few blocks at a time doesn't +# repeatedly rescan and recopy the buffered bytes. +stream_pull <- function(resp, n, splitter, max_size) { + cache <- resp$cache + queue <- cache$stream_queue + pos <- cache$stream_pos + + # Accumulate served slices in a list and flatten once at the end, so serving + # many blocks (e.g. `lines = Inf`) doesn't repeatedly recopy a growing vector. + serve <- list() + n_out <- 0L + repeat { + # Serve whatever is already queued. + available <- length(queue) - pos + 1L + if (available > 0L) { + take <- min(available, n - n_out) + serve[[length(serve) + 1L]] <- queue[pos:(pos + take - 1L)] + pos <- pos + take + n_out <- n_out + take } - - # If ending on \r, there's a special edge case here where if the - # next line begins with \n, that byte should be eaten. - if (utils::tail(line_bytes, 1) == 0x0D) { - resp$cache$resp_stream_oneline_eat_next_lf <- TRUE + if (n_out >= n) { + break } - # Use `resp$body` as the variable name so that if warn=TRUE, you get - # "incomplete final line found on 'resp$body'" as the warning message - `resp$body` <- line_bytes - line_con <- rawConnection(`resp$body`) - on.exit(close(line_con)) - - # readLines chomps the trailing newline. I assume this is desirable. - raw_text <- readLines(line_con, n = 1, warn = warn) - - # Use iconv to convert from whatever encoding is specified in the - # response header, to UTF-8 - return(iconv(raw_text, encoding, "UTF-8")) - } -} - -find_line_boundary <- function(buffer) { - if (length(buffer) == 0) { - return(NULL) - } - - # Look left 1 byte - right1 <- c(utils::tail(buffer, -1), 0x00) - - crlf <- buffer == 0x0D & right1 == 0x0A - cr <- buffer == 0x0D - lf <- buffer == 0x0A - - all <- which(crlf | cr | lf) - if (length(all) == 0) { - return(NULL) - } - - first <- all[[1]] - if (crlf[first]) { - return(first + 2) - } else { - return(first + 1) - } -} - -# Function to find the first double line ending in a buffer, or NULL if no -# double line ending is found -# -# Example: -# find_event_boundary(charToRaw("data: 1\n\nid: 12345")) -# Returns: -# list( -# matched = charToRaw("data: 1\n\n"), -# remaining = charToRaw("id: 12345") -# ) -find_event_boundary <- function(buffer) { - if (length(buffer) < 2) { - return(NULL) - } - - # leftX means look behind by X bytes. For example, left1[2] equals buffer[1]. - # Any attempt to read past the beginning of the buffer results in 0x00. - left1 <- c(0x00, utils::head(buffer, -1)) - left2 <- c(0x00, utils::head(left1, -1)) - left3 <- c(0x00, utils::head(left2, -1)) - - 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 - ) + # The queue is exhausted; combine any buffered bytes with a fresh read and + # split the next batch of blocks. We always reparse the buffered bytes (not + # just freshly read ones), because data may have been buffered by an earlier + # call that didn't find a complete block. - if (length(boundary_end) == 0) { - 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 - split_at -} - -# Splits a buffer into the part before `split_at`, and the part starting at -# `split_at`. It's possible for either of the returned parts to be zero-length -# (i.e. if `split_at` is 1 or length(buffer)+1). -split_buffer <- function(buffer, split_at) { - # Return a list with the event data and the remaining buffer - list( - matched = slice(buffer, end = split_at), - remaining = slice(buffer, start = split_at) - ) -} - -# @param max_size Maximum number of bytes to look for a boundary before throwing an error -# @param boundary_func A function that takes a raw vector and returns NULL if no -# boundary was detected, or one position PAST the end of the first boundary in -# 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 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) - - chunk_size <- min(max_size + 1, 1024) - - # Grab data left over from last resp_stream_sse() call (if any) - buffer <- resp$cache$push_back %||% raw() - resp$cache$push_back <- raw() - - if (resp_stream_show_buffer(resp)) { - log_stream(cli::rule("Buffer"), prefix = "* ") - print_buffer <- function(buf, label) { - log_stream( - label, - ": ", - paste(as.character(buf), collapse = " "), - prefix = "* " + stream_buffer <- cache$stream_buffer + if (length(stream_buffer) > max_size) { + stop_stream_size(max_size) + } + # Read up to one byte past the size limit - the extra byte lets us detect a + # buffer that has overflowed - but never more than one chunk at a time. + if (is.finite(max_size)) { + read_size <- min( + stream_chunk_bytes, + max(max_size - length(stream_buffer) + 1L, 1L) ) + } else { + read_size <- stream_chunk_bytes } - } else { - print_buffer <- function(buf, label) {} - } - - # Read chunks until we find an event or reach the end of input - repeat { - # Try to find an event boundary using the data we have - print_buffer(buffer, "Buffer to parse") - split_at <- boundary_func(buffer) - - if (!is.null(split_at)) { - result <- split_buffer(buffer, split_at) - # We found a complete event - print_buffer(result$matched, "Matched data") - print_buffer(result$remaining, "Remaining buffer") - resp$cache$push_back <- result$remaining - return(result$matched) + chunk <- resp$body$read(read_size) + buffer <- c(stream_buffer, chunk) + if (length(buffer) == 0L) { + break } - if (length(buffer) > max_size) { - # Keep the buffer in place, so that if the user tries resp_stream_sse - # again, they'll get the same error rather than reading the stream - # having missed a bunch of bytes. - resp$cache$push_back <- buffer - cli::cli_abort( - "Streaming read exceeded size limit of {max_size}", - class = "httr2_streaming_error" + if (length(chunk) > 0L && resp_stream_show_buffer(resp)) { + log_stream(cli::rule("Buffer"), prefix = "* ") + log_stream( + "Received chunk: ", + paste(as.character(chunk), collapse = " "), + prefix = "* " ) } - # We didn't have enough data. Attempt to read more, but don't let us exceed - # the max size by more than one byte; we do allow the one extra byte so we - # know to error. - chunk <- resp$body$read(min(chunk_size, max_size - length(buffer) + 1)) - print_buffer(chunk, "Received chunk") + # Divide the buffer into complete blocks at the splitter's boundaries. + splits <- splitter$find_boundaries(buffer) + if (length(splits) > 0L) { + starts <- c(1L, splits[-length(splits)]) + queue <- lapply(seq_along(splits), function(i) { + buffer[starts[i]:(splits[i] - 1L)] + }) + pos <- 1L + # Trailing bytes after the last boundary stay buffered for the next read. + cache$stream_buffer <- buffer[seq2( + splits[[length(splits)]], + length(buffer) + )] + # Checkpoint the freshly parsed queue before serving from it: if a later + # read trips the size limit, the errored call is retried and these blocks + # are served again rather than lost (their bytes are already consumed). + cache$stream_queue <- queue + cache$stream_pos <- pos + next + } - if (length(chunk) == 0) { + # No complete block in the buffer; keep it buffered for the next read. + cache$stream_buffer <- buffer + if (length(chunk) == 0L) { if (resp$body$is_complete()) { - # We've truly reached the end of the connection; no more data is coming - 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) - } + # The stream has ended; let the splitter flush any trailing bytes. + final <- splitter$finish(buffer) + if (length(final) > 0L) { + serve[[length(serve) + 1L]] <- final } - } 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) + cache$stream_buffer <- raw() } + # Either EOF, or no data currently available (non-blocking). + break } - - # More data was received; combine it with existing buffer and continue the - # loop to try parsing again - buffer <- c(buffer, chunk) - print_buffer(buffer, "Combined buffer") + # We read new bytes but still don't have a complete block; loop to read more. } -} -# https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation -parse_event <- function(event_data) { - if (is.raw(event_data)) { - # Streams must be decoded using the UTF-8 decode algorithm. - str_data <- rawToChar(event_data) - Encoding(str_data) <- "UTF-8" - } else { - # for testing - str_data <- event_data + # Drop the queue once it's been fully served, so its blocks can be freed. + if (pos > length(queue)) { + queue <- list() + pos <- 1L } + cache$stream_queue <- queue + cache$stream_pos <- pos - # The stream must then be parsed by reading everything line by line, with a - # U+000D CARRIAGE RETURN U+000A LINE FEED (CRLF) character pair, a single - # U+000A LINE FEED (LF) character not preceded by a U+000D CARRIAGE RETURN - # (CR) character, and a single U+000D CARRIAGE RETURN (CR) character not - # followed by a U+000A LINE FEED (LF) character being the ways in - # which a line can end. - lines <- strsplit(str_data, "\r\n|\r|\n")[[1]] - - # When a stream is parsed, a data buffer, an event type buffer, and a - # last event ID buffer must be associated with it. They must be initialized - # to the empty string. - data <- "" - type <- "" - last_id <- "" - - # If the line starts with a U+003A COLON character (:) - Ignore the line. - lines <- lines[!grepl("^:", lines)] - - # If the line contains a U+003A COLON character (:) - # * Collect the characters on the line before the first U+003A COLON - # character (:), and let field be that string. - # * Collect the characters on the line after the first U+003A COLON character - # (:), and let value be that string. If value starts with a U+0020 SPACE - # character, remove it from value. - m <- regexec("([^:]*)(: ?)?(.*)", lines) - matches <- regmatches(lines, m) - keys <- c("event", vapply(matches, function(x) x[2], character(1))) - values <- c("message", vapply(matches, function(x) x[4], character(1))) - - for (i in seq_along(matches)) { - key <- matches[[i]][2] - value <- matches[[i]][4] + unlist(serve, recursive = FALSE, use.names = FALSE) +} - if (key == "event") { - # Set the event type buffer to field value. - type <- value - } else if (key == "data") { - # Append the field value to the data buffer, then append a single - # U+000A LINE FEED (LF) character to the data buffer. - data <- paste0(data, value, "\n") - } else if (key == "id") { - # If the field value does not contain U+0000 NULL, then set the last - # event ID buffer to the field value. Otherwise, ignore the field. - last_id <- value +# A `StreamSplitter` describes a stream format for `stream_pull()`, which does +# the reading, splitting, and queueing. Subclasses (lines, server-sent events, +# AWS events) supply a `name` plus `find_boundaries()`, which takes a raw vector +# and returns an integer vector of split points: the position one past the end +# of each complete block. +StreamSplitter <- R6::R6Class( + "StreamSplitter", + public = list( + # The full name of the public reader (e.g. "resp_stream_lines()") this + # splitter backs. Used to detect mixing readers on one response. + name = NULL, + # nocov start: abstract, always overridden by a subclass. + find_boundaries = function(buffer) { + cli::cli_abort("Not implemented.", .internal = TRUE) + }, + # nocov end + # Emit any final blocks once the stream has ended with `remainder` bytes + # left over after the last complete block. The default discards a trailing + # partial block; line splitting overrides this to keep it. + finish = function(remainder) { + if (length(remainder) != 0L) { + cli::cli_warn("Premature end of input; ignoring final partial chunk") + } + list() } - } - - # If the data buffer is an empty string, set the data buffer and the event - # type buffer to the empty string and return. - if (data == "") { - return() - } - - # If the data buffer's last character is a U+000A LINE FEED (LF) character, - # then remove the last character from the data buffer. - if (grepl("\n$", data)) { - data <- substr(data, 1, nchar(data) - 1) - } - if (type == "") { - type <- "message" - } - - list( - type = type, - data = data, - id = last_id + ) +) + +# resp_stream_raw() reads bytes straight off the connection, so it never goes +# through stream_pull() and needs no splitting behavior. It's duck-typed rather +# than a StreamSplitter subclass: init_streaming_response() only needs a `name` +# (for reader tracking). +RawSplitter <- R6::R6Class( + "RawSplitter", + public = list(name = "resp_stream_raw()") +) + +stop_stream_size <- function(max_size, call = caller_env()) { + cli::cli_abort( + "Streaming read exceeded size limit of {max_size}", + class = "httr2_streaming_error", + call = call ) } # Helpers ---------------------------------------------------- -check_streaming_response <- function( +# Validate a streaming response, initialize its cache, and (when a `splitter` +# generator is supplied) construct and cache the splitter. The cached splitter +# both drives the reads and records which reader is in use, so attempting a +# second, different reader on the same response errors. Returns the splitter, +# so callers can write `splitter <- init_streaming_response(resp, SseSplitter)`. +init_streaming_response <- function( resp, + splitter = NULL, arg = caller_arg(resp), call = caller_env() ) { @@ -468,20 +282,34 @@ check_streaming_response <- function( if (!resp$body$is_open()) { cli::cli_abort("{.arg {arg}} has already been closed.", call = call) } -} -# isOpen doesn't work for two reasons: -# 1. It errors if con has been closed, rather than returning FALSE -# 2. If returns TRUE if con has been closed and a new connection opened -# -# So instead we retrieve the connection from its number and compare to the -# original connection. This works because connections have an undocumented -# external pointer. -isValid <- function(con) { - tryCatch( - identical(getConnection(con), con), - error = function(cnd) FALSE - ) + # Initialize the streaming cache so the read loop and stream_has_buffered() + # can read these fields without `%||%` guards. Only fills missing fields, so + # it preserves any bytes a caller has already pushed back. + cache <- resp$cache + cache$stream_buffer <- cache$stream_buffer %||% raw() + cache$stream_queue <- cache$stream_queue %||% list() + cache$stream_pos <- cache$stream_pos %||% 1L + + if (is.null(splitter)) { + return(invisible(NULL)) + } + + # Construct the splitter once per response and cache it. + cached <- cache$stream_splitter + if (is.null(cached)) { + cache$stream_splitter <- splitter$new() + return(invisible(cache$stream_splitter)) + } + if (!inherits(cached, splitter$classname)) { + used <- splitter$new()$name + cli::cli_abort( + "Can't use {used} after {cached$name} on the same response.", + class = "httr2_streaming_error", + call = call + ) + } + invisible(cached) } resp_stream_show_body <- function(resp) { diff --git a/R/test.R b/R/test.R index 3324e0e50..41005d8b4 100644 --- a/R/test.R +++ b/R/test.R @@ -20,9 +20,6 @@ request_test <- function(template = "/get", ...) { #' @export example_url <- function(path = "/") { check_installed("webfakes") - if (is_testing() && !interactive()) { - testthat::skip_on_covr() - } env_cache(the, "test_app", example_app()) the$test_app$url(path) } diff --git a/man/resp_stream_raw.Rd b/man/resp_stream_raw.Rd index 48dba2d6b..fca87b7ee 100644 --- a/man/resp_stream_raw.Rd +++ b/man/resp_stream_raw.Rd @@ -1,5 +1,6 @@ % Generated by roxygen2: do not edit by hand -% Please edit documentation in R/resp-stream.R, R/resp-stream-aws.R +% Please edit documentation in R/resp-stream.R, R/resp-stream-lines.R, +% R/resp-stream-sse.R, R/resp-stream-aws.R \name{resp_stream_raw} \alias{resp_stream_raw} \alias{resp_stream_lines} @@ -11,7 +12,7 @@ \usage{ resp_stream_raw(resp, kb = 32) -resp_stream_lines(resp, lines = 1, max_size = Inf, warn = TRUE) +resp_stream_lines(resp, lines = 1, max_size = Inf, warn = deprecated()) resp_stream_sse(resp, max_size = Inf) @@ -28,11 +29,14 @@ resp_stream_is_complete(resp) \item{lines}{The maximum number of lines to return at once.} -\item{max_size}{The maximum number of bytes to buffer; once this number of -bytes has been exceeded without a line/event boundary, an error is thrown.} +\item{max_size}{The maximum number of bytes to buffer while waiting for a +line or event boundary; if exceeded, an error is thrown. This limit is +approximate: to spot a boundary httr2 may buffer a handful of bytes beyond +\code{max_size} (e.g. the bytes of the delimiter itself).} -\item{warn}{Like \code{\link[=readLines]{readLines()}}: warn if the connection ends without a final -EOL.} +\item{warn}{\ifelse{html}{\href{https://lifecycle.r-lib.org/articles/stages.html#deprecated}{\figure{lifecycle-deprecated.svg}{options: alt='[Deprecated]'}}}{\strong{[Deprecated]}} \code{resp_stream_lines()} no longer +warns when the connection ends without a final EOL, so this argument is +ignored.} \item{...}{Not used; included for compatibility with generic.} } diff --git a/tests/testthat/_snaps/resp-stream-aws.md b/tests/testthat/_snaps/resp-stream-aws.md index 1a714124b..a011ac2bd 100644 --- a/tests/testthat/_snaps/resp-stream-aws.md +++ b/tests/testthat/_snaps/resp-stream-aws.md @@ -1,10 +1,31 @@ # unknown header triggers error Code - parse_aws_event(bytes) + parse_aws_event(aws_event(aws_header("foo", "unknown"))) Condition Error in `type_enum()`: ! Unsupported type 255. i This is an internal error that was detected in the httr2 package. Please report it at with a reprex () and the full backtrace. +# parse_aws_event() checks the prelude length + + Code + parse_aws_event(as.raw(1:10)) + Condition + Error in `parse_aws_event()`: + ! AWS event metadata doesn't match supplied bytes + i This is an internal error that was detected in the httr2 package. + Please report it at with a reprex () and the full backtrace. + +# verbosity = 3 shows aws events + + Code + . <- resp_stream_aws(resp) + Output + * -- Buffer ---------------------------------------------------------------------- + * Received chunk: 00 00 00 1a 00 00 00 0a af a7 b9 54 03 66 6f 6f 07 00 03 62 61 72 c0 c0 6d f5 + << foo: bar + << "" + + diff --git a/tests/testthat/_snaps/resp-stream-lines.md b/tests/testthat/_snaps/resp-stream-lines.md new file mode 100644 index 000000000..17ceed09b --- /dev/null +++ b/tests/testthat/_snaps/resp-stream-lines.md @@ -0,0 +1,8 @@ +# resp_stream_lines(warn) is deprecated unless FALSE + + Code + . <- resp_stream_lines(resp, warn = TRUE) + Condition + Warning: + The `warn` argument of `resp_stream_lines()` is deprecated as of httr2 1.2.3. + diff --git a/tests/testthat/_snaps/resp-stream-sse.md b/tests/testthat/_snaps/resp-stream-sse.md new file mode 100644 index 000000000..aba5a5559 --- /dev/null +++ b/tests/testthat/_snaps/resp-stream-sse.md @@ -0,0 +1,28 @@ +# can determine if incomplete data is complete + + Code + expect_equal(resp_stream_sse(con), NULL) + Condition + Warning: + Premature end of input; ignoring final partial chunk + +# verbosity = 3 shows raw sse events + + Code + . <- resp_stream_sse(resp) + Output + * -- Buffer ---------------------------------------------------------------------- + * Received chunk: 3a 20 63 6f 6d 6d 65 6e 74 0a 0a 64 61 74 61 3a 20 31 0a 0a + * -- Raw server sent event ------------------------------------------------------- + * : comment + * + * + * -- Raw server sent event ------------------------------------------------------- + * data: 1 + * + * + << type: message + << data: 1 + << id: + + diff --git a/tests/testthat/_snaps/resp-stream.md b/tests/testthat/_snaps/resp-stream.md index 51767f033..ed039260d 100644 --- a/tests/testthat/_snaps/resp-stream.md +++ b/tests/testthat/_snaps/resp-stream.md @@ -1,37 +1,68 @@ -# can determine if incomplete data is complete +# can't read from a closed connection Code - expect_equal(resp_stream_sse(con), NULL) + resp_stream_raw(resp, 1) Condition - Warning: - Premature end of input; ignoring final partial chunk + Error in `resp_stream_raw()`: + ! `resp` has already been closed. -# can't read from a closed connection +# streaming functions require a streaming response Code - resp_stream_raw(resp, 1) + resp_stream_raw(response()) Condition Error in `resp_stream_raw()`: + ! `resp` must be a streaming HTTP response object, not a object. + +# resp_stream_raw() validates kb + + Code + resp_stream_raw(resp, kb = -1) + Condition + Error in `resp_stream_raw()`: + ! `kb` must be a number larger than or equal to 0, not the number -1. + +--- + + Code + resp_stream_raw(resp, kb = Inf) + Condition + Error in `resp_stream_raw()`: + ! `kb` must be a number, not `Inf`. + +# resp_stream_is_complete() requires an open streaming response + + Code + resp_stream_is_complete(response()) + Condition + Error in `resp_stream_is_complete()`: + ! `resp` must be a streaming HTTP response object, not a object. + +--- + + Code + resp_stream_is_complete(resp) + Condition + Error in `resp_stream_is_complete()`: ! `resp` has already been closed. -# verbosity = 2 streams request bodies +# streaming responses use only one reader Code - stream_all(req, resp_stream_lines, 1) - Output - << line 1 - << line 2 + resp_stream_raw(resp) + Condition + Error in `resp_stream_raw()`: + ! Can't use resp_stream_raw() after resp_stream_lines() on the same response. + +# StreamSplitter$finish() discards a trailing partial block with a warning + Code - stream_all(req, resp_stream_raw, 5 / 1024) - Output - << Streamed 5 bytes - - << Streamed 5 bytes - - << Streamed 4 bytes - + out <- s$finish(charToRaw("b")) + Condition + Warning: + Premature end of input; ignoring final partial chunk -# verbosity = 3 shows buffer info +# verbosity = 3 logs the buffered chunk Code while (!resp_stream_is_complete(con)) { @@ -39,45 +70,24 @@ } Output * -- Buffer ---------------------------------------------------------------------- - * Buffer to parse: * Received chunk: 6c 69 6e 65 20 31 0a 6c 69 6e 65 20 32 0a - * Combined buffer: 6c 69 6e 65 20 31 0a 6c 69 6e 65 20 32 0a - * Buffer to parse: 6c 69 6e 65 20 31 0a 6c 69 6e 65 20 32 0a - * Matched data: 6c 69 6e 65 20 31 0a - * Remaining buffer: 6c 69 6e 65 20 32 0a << line 1 - * -- Buffer ---------------------------------------------------------------------- - * Buffer to parse: 6c 69 6e 65 20 32 0a - * Matched data: 6c 69 6e 65 20 32 0a - * Remaining buffer: << line 2 -# verbosity = 3 shows raw sse events +# verbosity = 2 streams request bodies Code - . <- resp_stream_sse(resp) + stream_all(req, resp_stream_lines, 1) Output - * -- Buffer ---------------------------------------------------------------------- - * Buffer to parse: - * Received chunk: 3a 20 63 6f 6d 6d 65 6e 74 0a 0a 64 61 74 61 3a 20 31 0a 0a - * Combined buffer: 3a 20 63 6f 6d 6d 65 6e 74 0a 0a 64 61 74 61 3a 20 31 0a 0a - * Buffer to parse: 3a 20 63 6f 6d 6d 65 6e 74 0a 0a 64 61 74 61 3a 20 31 0a 0a - * Matched data: 3a 20 63 6f 6d 6d 65 6e 74 0a 0a - * Remaining buffer: 64 61 74 61 3a 20 31 0a 0a - * -- Raw server sent event ------------------------------------------------------- - * : comment - * - * - * -- Buffer ---------------------------------------------------------------------- - * Buffer to parse: 64 61 74 61 3a 20 31 0a 0a - * Matched data: 64 61 74 61 3a 20 31 0a 0a - * Remaining buffer: - * -- Raw server sent event ------------------------------------------------------- - * data: 1 - * - * - << type: message - << data: 1 - << id: + << line 1 + << line 2 + Code + stream_all(req, resp_stream_raw, 5 / 1024) + Output + << Streamed 5 bytes + + << Streamed 5 bytes + + << Streamed 4 bytes diff --git a/tests/testthat/helper-aws.R b/tests/testthat/helper-aws.R new file mode 100644 index 000000000..1c30a301d --- /dev/null +++ b/tests/testthat/helper-aws.R @@ -0,0 +1,63 @@ +# Helpers for building AWS event-stream messages in tests. + +# Decode a hex string to raw. Only used to express opaque reference vectors +# captured from another implementation; build events with aws_event() instead. +hex_to_raw <- function(x) { + x <- gsub("(\\s|\n)+", "", x) + pairs <- substring(x, seq(1, nchar(x), by = 2), seq(2, nchar(x), by = 2)) + as.raw(strtoi(pairs, 16L)) +} + +# A big-endian integer in `size` raw bytes. Handles values beyond +# .Machine$integer.max (unlike writeBin()) and encodes negatives as two's +# complement, so it serves both unsigned lengths and signed header values. +aws_uint <- function(x, size) { + x <- as.numeric(x) + out <- raw(size) + for (i in seq_len(size)) { + out[[size - i + 1L]] <- as.raw(x %% 256) + x <- x %/% 256 + } + out +} + +# CRC32 of `bytes` as 4 big-endian raw bytes, matching the AWS event-stream +# framing. +aws_crc <- function(bytes) { + hex <- digest::digest(bytes, algo = "crc32", serialize = FALSE) + as.raw(strtoi(substring(hex, c(1L, 3L, 5L, 7L), c(2L, 4L, 6L, 8L)), 16L)) +} + +# A single header: a `name`, a `type` (matching the AWS event-stream spec), and +# a `value` to encode. +aws_header <- function(name, type, value = NULL) { + body <- switch( + type, + true = list(tag = 0L, bytes = raw()), + false = list(tag = 1L, bytes = raw()), + byte = list(tag = 2L, bytes = aws_uint(value, 1L)), + short = list(tag = 3L, bytes = aws_uint(value, 2L)), + integer = list(tag = 4L, bytes = aws_uint(value, 4L)), + long = list(tag = 5L, bytes = aws_uint(value, 8L)), + bytes = list(tag = 6L, bytes = c(aws_uint(length(value), 2L), value)), + string = list( + tag = 7L, + bytes = c(aws_uint(nchar(value), 2L), charToRaw(value)) + ), + timestamp = list(tag = 8L, bytes = aws_uint(value, 8L)), + uuid = list(tag = 9L, bytes = value), + unknown = list(tag = 255L, bytes = raw()), + cli::cli_abort("Unknown header type {.val {type}}.") + ) + c(aws_uint(nchar(name), 1L), charToRaw(name), as.raw(body$tag), body$bytes) +} + +# A complete event wrapping raw `headers` and a raw `body` in the AWS +# event-stream framing, computing the prelude/header lengths and both CRCs. +aws_event <- function(headers = raw(), body = raw()) { + total <- 12L + length(headers) + length(body) + 4L + prelude <- c(aws_uint(total, 4L), aws_uint(length(headers), 4L)) + prelude <- c(prelude, aws_crc(prelude)) # prelude CRC over the first 8 bytes + message <- c(prelude, headers, body) + c(message, aws_crc(message)) # message CRC over everything before it +} diff --git a/tests/testthat/helper-resp-stream.R b/tests/testthat/helper-resp-stream.R new file mode 100644 index 000000000..df8081ce4 --- /dev/null +++ b/tests/testthat/helper-resp-stream.R @@ -0,0 +1,5 @@ +local_streaming_response <- function(data, frame = parent.frame()) { + resp <- response(body = StreamingBody$new(rawConnection(data, "rb"))) + withr::defer(close(resp), envir = frame) + resp +} diff --git a/tests/testthat/helper-sync.R b/tests/testthat/helper-sync.R index 49d4ff954..f7c46a25c 100644 --- a/tests/testthat/helper-sync.R +++ b/tests/testthat/helper-sync.R @@ -64,7 +64,7 @@ wait_for_http_data <- function(resp, timeout_s = 5) { while (as.double(Sys.time()) < deadline) { chunk <- resp$body$read(256) if (length(chunk) > 0) { - resp$cache$push_back <- c(resp$cache$push_back, chunk) + resp$cache$stream_buffer <- c(resp$cache$stream_buffer, chunk) return(invisible(TRUE)) } @@ -92,7 +92,7 @@ wait_for_complete <- function(resp, timeout_s = 5) { repeat { chunk <- resp$body$read(256) if (length(chunk) > 0) { - resp$cache$push_back <- c(resp$cache$push_back, chunk) + resp$cache$stream_buffer <- c(resp$cache$stream_buffer, chunk) } if (resp$body$is_complete()) { diff --git a/tests/testthat/helper-webfakes.R b/tests/testthat/helper-webfakes.R index e79d2e6b8..a50fa8a61 100644 --- a/tests/testthat/helper-webfakes.R +++ b/tests/testthat/helper-webfakes.R @@ -1,11 +1,6 @@ local_app_request <- function(fun, method = "get", frame = parent.frame()) { # sometimes fails on CRAN and we don't need the hassle skip_on_cran() - # Works interactively (useful for manaul coverage checking) - # but not in separate process - if (!interactive()) { - skip_on_covr() - } app <- webfakes::new_app() diff --git a/tests/testthat/test-resp-stream-aws.R b/tests/testthat/test-resp-stream-aws.R index e932f1015..64fceca24 100644 --- a/tests/testthat/test-resp-stream-aws.R +++ b/tests/testthat/test-resp-stream-aws.R @@ -1,85 +1,188 @@ -# Tests copied from -# https://github.com/lifion/lifion-aws-event-stream/blob/develop/lib/index.test.js -# https://github.com/lifion/lifion-aws-event-stream/blob/develop/lib/index.test.json - test_that("can parse empty object", { - bytes <- hex_to_raw("000000100000000005c248eb7d98c8ff") expect_equal( - parse_aws_event(bytes), + parse_aws_event(aws_event()), list(headers = list(), body = "") ) }) + test_that("can return various types of header", { - bytes <- hex_to_raw("0000001500000001ba25f70d03666f6f013aa3e0d6") - expect_equal(parse_aws_event(bytes)$headers, list(foo = FALSE)) + headers <- function(...) parse_aws_event(aws_event(aws_header(...)))$headers + + expect_equal(headers("foo", "false"), list(foo = FALSE)) + expect_equal(headers("foo", "true"), list(foo = TRUE)) + expect_equal(headers("foo", "bytes", as.raw(1:5)), list(foo = as.raw(1:5))) + expect_equal(headers("foo", "string", "bar"), list(foo = "bar")) + + # byte, short, and integer are signed (two's complement) + expect_equal(headers("foo", "byte", 127), list(foo = 127)) + expect_equal(headers("foo", "byte", -1), list(foo = -1)) + expect_equal(headers("foo", "short", -2), list(foo = -2)) + expect_equal(headers("foo", "integer", -3), list(foo = -3)) + + # long and timestamp are 64-bit integers, returned as bit64::integer64 (see + # the reference test below for a non-trivial long value) + expect_equal( + headers("foo", "timestamp", 0), + list(foo = structure(0, class = "integer64")) + ) + + # UUID is returned as a hex string + uuid <- as.raw(1:16) + expect_equal(headers("foo", "uuid", uuid), list(foo = raw_to_hex(uuid))) +}) + - bytes <- hex_to_raw("0000001500000001ba25f70d03666f6f004da4d040") - expect_equal(parse_aws_event(bytes)$headers, list(foo = TRUE)) +test_that("unknown header triggers error", { + expect_snapshot( + parse_aws_event(aws_event(aws_header("foo", "unknown"))), + error = TRUE + ) +}) + +test_that("parse_aws_event() checks the prelude length", { + expect_snapshot(parse_aws_event(as.raw(1:10)), error = TRUE) +}) - # byte - bytes <- hex_to_raw("0000001600000001fd858ddd03666f6f02ffa44bfd93") - expect_equal(parse_aws_event(bytes)$headers, list(foo = 255)) +test_that("can read aws events one at a time", { + # Two empty-object events back to back. + event <- aws_event() + req <- local_app_request(function(req, res) { + res$send_chunk(event) + res$send_chunk(event) + }) + resp <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp)) - # short - bytes <- hex_to_raw("0000001700000001c0e5a46d03666f6f03fffff3b59291") - expect_equal(parse_aws_event(bytes)$headers, list(foo = 65535)) + expect_equal(resp_stream_aws(resp), list(headers = list(), body = "")) + expect_equal(resp_stream_aws(resp), list(headers = list(), body = "")) + expect_equal(resp_stream_aws(resp), NULL) +}) - # integer - bytes <- hex_to_raw("00000019000000017fd51a0c03666f6f04ffffffff853b65dd") - expect_equal(parse_aws_event(bytes)$headers, list(foo = 4294967295)) +test_that("max_size counts the buffered event bytes, approximately", { + data <- aws_event() + resp1 <- local_streaming_response(data) - # long - bytes <- hex_to_raw( - "0000001d000000018a55bccc03666f6f050000ffffffffffff6b03c255" + expect_equal( + resp_stream_aws(resp1, max_size = length(data)), + list(headers = list(), body = "") ) - expected <- structure(1.390671161567e-309, class = "integer64") - expect_equal(parse_aws_event(bytes)$headers, list(foo = expected)) - # byte array - bytes <- hex_to_raw( - "0000001c00000001b735957c03666f6f0600050102030405cdda4038" + resp2 <- local_streaming_response(data) + expect_error( + resp_stream_aws(resp2, max_size = length(data) - 2L), + class = "httr2_streaming_error" ) - expect_equal(parse_aws_event(bytes)$headers, list(foo = as.raw(1:5))) +}) - # character - bytes <- hex_to_raw("0000001a00000001387560dc03666f6f0700036261725bb3cecf") - expect_equal(parse_aws_event(bytes)$headers, list(foo = "bar")) +test_that("verbosity = 3 shows aws events", { + # A single event with a "foo: bar" header and empty body. + event <- aws_event(aws_header("foo", "string", "bar")) + req <- local_app_request(function(req, res) { + res$send_chunk(event) + }) - # UUID - bytes <- hex_to_raw( - "00000025000000011b044f8b03666f6f093bfdac5cfe6c402983bfc1de7819f5316056148a" + expect_output(resp <- req_perform_connection(req, verbosity = 3)) + withr::defer(close(resp)) + expect_snapshot( + . <- resp_stream_aws(resp), + transform = transform_verbose_response ) +}) + +test_that("find_aws_event_boundaries splits a buffer into complete events", { + event <- aws_event() + + expect_equal(find_aws_event_boundaries(event), 17) + expect_equal(find_aws_event_boundaries(c(event, event)), c(17, 33)) expect_equal( - parse_aws_event( - bytes - )$headers, - list(foo = "3bfdac5cfe6c402983bfc1de7819f531") + find_aws_event_boundaries(c(event, event, event)), + c(17, 33, 49) ) }) -test_that("unknown header triggers error", { - bytes <- hex_to_raw("0000001500000001ba25f70d03666f6fff60a63fcd") - expect_snapshot(parse_aws_event(bytes), error = TRUE) +test_that("find_aws_event_boundaries ignores incomplete trailing events", { + event <- aws_event() + + # Nothing to split + expect_equal(find_aws_event_boundaries(raw()), double()) + # Fewer than 16 bytes can't be a complete event + expect_equal(find_aws_event_boundaries(event[1:10]), double()) + # Trailing partial event is excluded + expect_equal(find_aws_event_boundaries(c(event, event[1:8])), 17) + # Event claiming more bytes than are available is excluded + big <- event + big[1:4] <- aws_uint(256, 4L) + expect_equal(find_aws_event_boundaries(c(event, big)), 17) }) test_that("json content type automatically parsed", { - bytes <- hex_to_raw( - " - 000001c20000005bc1123f0b0b3a6576656e742d74797065070015537562736372696265546f - 53686172644576656e740d3a636f6e74656e742d747970650700106170706c69636174696f6e - 2f6a736f6e0d3a6d6573736167652d747970650700056576656e747b22436f6e74696e756174 - 696f6e53657175656e63654e756d626572223a22343935383836333037393634323435313235 - 3936363136333437353239313133373435393934373336323937343734373039373832353330 - 222c224d696c6c6973426568696e644c6174657374223a302c225265636f726473223a5b7b22 - 417070726f78696d6174654172726976616c54696d657374616d70223a312e35333831363032 - 313936333645392c2244617461223a225632567a62475635222c22456e6372797074696f6e54 - 797065223a6e756c6c2c22506172746974696f6e4b6579223a2231306463633930322d633839 - 632d343036372d623433362d303566383863306662356566222c2253657175656e63654e756d - 626572223a223439353838363330373936343234353132353936363136333437353239313133 - 373435393934373336323937343734373039373832353330227d5d7dd84c02f3 - " - ) - parsed <- parse_aws_event(bytes) - expect_type(parsed$body, "list") + event <- aws_event( + c( + aws_header(":event-type", "string", "SubscribeToShardEvent"), + aws_header(":content-type", "string", "application/json"), + aws_header(":message-type", "string", "event") + ), + body = charToRaw('{"records": []}') + ) + parsed <- parse_aws_event(event) + expect_equal(parsed$body, list(records = list())) +}) + +# aws_event() ------------------------------------------------------------------ + +test_that("aws_event() produces spec-valid bytes, including CRCs", { + # Using values from a reference implementation at + # https://github.com/lifion/lifion-aws-event-stream + expect_equal(aws_event(), hex_to_raw("000000100000000005c248eb7d98c8ff")) +}) + +test_that("aws_event() agrees with the reference implementation", { + # Messages captured from the lifion JS reference implementation: + # https://github.com/lifion/lifion-aws-event-stream. These vectors encode a + # non-spec header length (always 1), so their bytes differ from aws_event()'s, + # but a reference message and the equivalent aws_event() must decode the same. + agrees <- function(reference, header) { + expect_equal( + parse_aws_event(hex_to_raw(reference)), + parse_aws_event(aws_event(header)) + ) + } + + agrees( + "0000001500000001ba25f70d03666f6f013aa3e0d6", + aws_header("foo", "false") + ) + agrees( + "0000001500000001ba25f70d03666f6f004da4d040", + aws_header("foo", "true") + ) + agrees( + "0000001600000001fd858ddd03666f6f02ffa44bfd93", + aws_header("foo", "byte", -1) # 0xff + ) + agrees( + "0000001700000001c0e5a46d03666f6f03fffff3b59291", + aws_header("foo", "short", -1) # 0xffff + ) + agrees( + "00000019000000017fd51a0c03666f6f04ffffffff853b65dd", + aws_header("foo", "integer", -1) # 0xffffffff + ) + agrees( + "0000001d000000018a55bccc03666f6f050000ffffffffffff6b03c255", + aws_header("foo", "long", 281474976710655) # 0x0000ffffffffffff + ) + agrees( + "0000001a00000001387560dc03666f6f0700036261725bb3cecf", + aws_header("foo", "string", "bar") + ) + agrees( + "0000001c00000001b735957c03666f6f0600050102030405cdda4038", + aws_header("foo", "bytes", as.raw(1:5)) + ) + agrees( + "00000025000000011b044f8b03666f6f093bfdac5cfe6c402983bfc1de7819f5316056148a", + aws_header("foo", "uuid", hex_to_raw("3bfdac5cfe6c402983bfc1de7819f531")) + ) }) diff --git a/tests/testthat/test-resp-stream-lines.R b/tests/testthat/test-resp-stream-lines.R new file mode 100644 index 000000000..8836a0299 --- /dev/null +++ b/tests/testthat/test-resp-stream-lines.R @@ -0,0 +1,118 @@ +test_that("decodes the response encoding and joins LF and CRLF lines", { + # Lines are decoded with the response encoding (here Shift_JIS) and split on + # both LF and CRLF, including a CRLF straddling two reads ("split crlf\r" then + # "\n"); a final line without a terminator is returned at EOF. + req <- local_app_request(function(req, res) { + res$set_header("Content-Type", "text/plain; charset=Shift_JIS") + res$send_chunk(as.raw(c(0x82, 0xA0, 0x0A))) + res$send_chunk("crlf\r\n") + res$send_chunk("lf\n") + res$send_chunk("half line/") + res$send_chunk("other half\n") + res$send_chunk("split crlf\r") + res$send_chunk("\nanother line\n") + res$send_chunk("eof without line ending") + }) + resp <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp)) + + expected_values <- c( + "\u3042", + "crlf", + "lf", + "half line/other half", + "split crlf", + "another line" + ) + + for (expected in expected_values) { + rlang::inject(expect_equal(resp_stream_lines(resp), !!expected)) + } + expect_equal(resp_stream_lines(resp), "eof without line ending") +}) + +test_that("requesting zero lines returns an empty vector", { + req <- local_app_request(function(req, res) res$send_chunk("a\n")) + resp <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp)) + expect_equal(resp_stream_lines(resp, 0), character()) +}) + +test_that("max_size is approximate, counting the buffered delimiter bytes", { + # A line whose content is exactly max_size is returned: we read one byte past + # max_size, enough to capture a single LF terminator. + resp <- local_streaming_response(c(rep(as.raw(0x61), 10), as.raw(0x0A))) + expect_equal(resp_stream_lines(resp, max_size = 10), strrep("a", 10)) + + # A two-byte CRLF tips the buffer one byte over, so the same line needs a + # slightly larger max_size. + crlf <- c(rep(as.raw(0x61), 10), as.raw(c(0x0D, 0x0A))) + resp <- local_streaming_response(crlf) + expect_error( + resp_stream_lines(resp, max_size = 10), + class = "httr2_streaming_error" + ) + expect_equal(resp_stream_lines(resp, max_size = 11), strrep("a", 10)) +}) + +test_that("max_size counts an incomplete delimiter at EOF as content", { + resp <- local_streaming_response(c(rep(as.raw(0x61), 10), as.raw(0x0D))) + + expect_error( + resp_stream_lines(resp, max_size = 10), + class = "httr2_streaming_error" + ) +}) + +test_that("resp_stream_lines(warn) is deprecated unless FALSE", { + req <- local_app_request(function(req, res) res$send_chunk("a\n")) + resp <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp)) + + # warn = FALSE already requested silence, so it's accepted quietly. + expect_no_warning(. <- resp_stream_lines(resp, warn = FALSE)) + # Any other value is deprecated. + expect_snapshot(. <- resp_stream_lines(resp, warn = TRUE)) +}) + +test_that("LineSplitter flushes a trailing line as a raw block", { + s <- LineSplitter$new() + # Nothing buffered: nothing to flush. + expect_equal(s$finish(raw()), list()) + # Trailing bytes are emitted as a final block. + expect_equal(s$finish(charToRaw("tail")), list(charToRaw("tail"))) + # A bare CR is ordinary content, even at end-of-stream. + expect_equal(s$finish(charToRaw("tail\r")), list(charToRaw("tail\r"))) +}) + +test_that("LineSplitter boundaries fall one past each LF", { + # The block-splitting itself lives in stream_pull(); here we just check where + # lines end. We only split on LF, so CRLF and bare CR need no special casing. + s <- LineSplitter$new() + + expect_equal(s$find_boundaries(charToRaw("a\nb\r\nc\n")), c(3L, 6L, 8L)) + # blank lines produce adjacent boundaries + expect_equal(s$find_boundaries(charToRaw("a\n\nb\n")), c(3L, 4L, 6L)) + # a bare CR (not followed by LF) is not a terminator + expect_equal(s$find_boundaries(charToRaw("a\rbcd")), integer()) +}) + +test_that("stream_parse_lines() decodes blocks and strips LF/CRLF terminators", { + blocks <- list( + charToRaw("a\n"), + charToRaw("b\r\n"), + charToRaw("\n"), + charToRaw("tail") + ) + expect_equal(stream_parse_lines(blocks, "UTF-8"), c("a", "b", "", "tail")) + + # honors the encoding, and a bare CR stays as content + expect_equal( + stream_parse_lines(list(as.raw(c(0x82, 0xA0, 0x0A))), "Shift_JIS"), + "あ" + ) + expect_equal(stream_parse_lines(list(charToRaw("a\rb\n")), "UTF-8"), "a\rb") + + # no blocks gives an empty vector + expect_equal(stream_parse_lines(list(), "UTF-8"), character()) +}) diff --git a/tests/testthat/test-resp-stream-sse.R b/tests/testthat/test-resp-stream-sse.R new file mode 100644 index 000000000..be8a8c880 --- /dev/null +++ b/tests/testthat/test-resp-stream-sse.R @@ -0,0 +1,180 @@ +test_that("can determine if incomplete data is complete", { + req <- local_app_request(function(req, res) { + res$send_chunk("data: 1\n\n") + res$send_chunk("data: ") + }) + + con <- req |> req_perform_connection(blocking = TRUE) + withr::defer(close(con)) + + expect_equal( + resp_stream_sse(con, 10), + list(type = "message", data = "1", id = "") + ) + expect_snapshot(expect_equal(resp_stream_sse(con), NULL)) + expect_true(resp_stream_is_complete(con)) +}) + +test_that("can feed sse events one at a time", { + req <- local_app_request(function(req, res) { + for (i in 1:3) { + res$send_chunk(sprintf("data: %s\n\n", i)) + } + }) + resp <- req_perform_connection(req) + withr::defer(close(resp)) + + expect_equal( + resp_stream_sse(resp), + list(type = "message", data = "1", id = "") + ) + expect_equal( + resp_stream_sse(resp), + list(type = "message", data = "2", id = "") + ) + resp_stream_sse(resp) + + expect_equal(resp_stream_sse(resp), NULL) +}) + +test_that("max_size counts the buffered event bytes, delimiter included", { + # The event with its "\r\n\r\n" delimiter is 11 bytes; max_size limits the + # buffered bytes (delimiter included), approximately. + data <- charToRaw("data: 1\r\n\r\n") + resp1 <- local_streaming_response(data) + + expect_equal( + resp_stream_sse(resp1, max_size = 10), + list(type = "message", data = "1", id = "") + ) + + resp2 <- local_streaming_response(data) + expect_error( + resp_stream_sse(resp2, max_size = 9), + class = "httr2_streaming_error" + ) +}) + +test_that("ignores events with no data", { + req <- local_app_request(function(req, res) { + res$send_chunk(": comment\n\n") + res$send_chunk("data: 1\n\n") + }) + resp <- req_perform_connection(req) + withr::defer(close(resp)) + + expect_equal( + resp_stream_sse(resp), + list(type = "message", data = "1", id = "") + ) +}) + +test_that("sse always interprets data as UTF-8", { + req <- local_app_request(function(req, res) { + res$send_chunk("data: \xE3\x81\x82\r\n\r\n") + }) + + # Data is decoded as UTF-8 regardless of the locale. + withr::local_locale(LC_CTYPE = "C") + resp <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp)) + + out <- resp_stream_sse(resp) + + s <- "\xE3\x81\x82" + Encoding(s) <- "UTF-8" + expect_equal(out, list(type = "message", data = s, id = "")) + expect_equal(Encoding(out$data), "UTF-8") +}) + +test_that("verbosity = 3 shows raw sse events", { + req <- local_app_request(function(req, res) { + res$send_chunk(": comment\n\n") + res$send_chunk("data: 1\n\n") + }) + + expect_output(resp <- req_perform_connection(req, verbosity = 3)) + withr::defer(close(resp)) + expect_snapshot( + . <- resp_stream_sse(resp), + transform = transform_verbose_response + ) +}) + +test_that("has a working find_event_boundaries", { + # Splits the buffer at the first boundary into the matched event and the + # remaining bytes. + boundary_test <- function(x, matched, remaining) { + buffer <- charToRaw(x) + splits <- find_event_boundaries(buffer) + result <- if (length(splits) == 0) { + NULL + } else { + split_at <- splits[[1]] + list( + matched = slice(buffer, end = split_at), + remaining = slice(buffer, start = split_at) + ) + } + expect_identical( + result, + list(matched = charToRaw(matched), remaining = charToRaw(remaining)) + ) + } + + # Basic matches + boundary_test("\r\r", matched = "\r\r", remaining = "") + boundary_test("\n\n", matched = "\n\n", remaining = "") + boundary_test("\r\n\r\n", matched = "\r\n\r\n", remaining = "") + boundary_test("a\r\r", matched = "a\r\r", remaining = "") + boundary_test("a\n\n", matched = "a\n\n", remaining = "") + boundary_test("a\r\n\r\n", matched = "a\r\n\r\n", remaining = "") + boundary_test("\r\ra", matched = "\r\r", remaining = "a") + boundary_test("\n\na", matched = "\n\n", remaining = "a") + boundary_test("\r\n\r\na", matched = "\r\n\r\n", remaining = "a") + + # Matches the first boundary found + boundary_test("\r\r\r", matched = "\r\r", remaining = "\r") + boundary_test("\r\r\r\r", matched = "\r\r", remaining = "\r\r") + boundary_test("\n\n\r\r", matched = "\n\n", remaining = "\r\r") + boundary_test("\r\r\n\n", matched = "\r\r", remaining = "\n\n") + + # Finds every boundary in the buffer + expect_equal(find_event_boundaries(charToRaw("a\n\nb\n\n")), c(4L, 7L)) + expect_equal( + find_event_boundaries(charToRaw("a\r\n\r\nb\r\n\r\n")), + c(6L, 11L) + ) + + # Non-matches + expect_length(find_event_boundaries(charToRaw("\n\r\n\r")), 0) + expect_length(find_event_boundaries(charToRaw("hello\ngoodbye\n")), 0) + expect_length(find_event_boundaries(charToRaw("")), 0) + expect_length(find_event_boundaries(charToRaw("1")), 0) + expect_length(find_event_boundaries(charToRaw("12")), 0) + expect_length(find_event_boundaries(charToRaw("\r\n\r")), 0) +}) + +# parse_event ---------------------------------------------------------------- + +test_that("event with no data returns NULL", { + expect_null(parse_event("")) + expect_null(parse_event(":comment")) + expect_null(parse_event("id: 1")) + + expect_equal(parse_event("data: ")$data, "") + expect_equal(parse_event("data")$data, "") +}) + +test_that("examples from spec work", { + event <- parse_event("data: YHOO\ndata: +2\ndata: 10") + expect_equal(event$type, "message") + expect_equal(event$data, "YHOO\n+2\n10") +}) + +test_that("event and id fields are parsed", { + event <- parse_event("event: ping\ndata: x\nid: 42") + expect_equal(event$type, "ping") + expect_equal(event$data, "x") + expect_equal(event$id, "42") +}) diff --git a/tests/testthat/test-resp-stream.R b/tests/testthat/test-resp-stream.R index 6eff4f36b..fd3cfb905 100644 --- a/tests/testthat/test-resp-stream.R +++ b/tests/testthat/test-resp-stream.R @@ -36,23 +36,6 @@ test_that("can determine if a stream is complete (non-blocking)", { expect_true(resp_stream_is_complete(resp)) }) -test_that("can determine if incomplete data is complete", { - req <- local_app_request(function(req, res) { - res$send_chunk("data: 1\n\n") - res$send_chunk("data: ") - }) - - con <- req |> req_perform_connection(blocking = TRUE) - withr::defer(close(con)) - - expect_equal( - resp_stream_sse(con, 10), - list(type = "message", data = "1", id = "") - ) - expect_snapshot(expect_equal(resp_stream_sse(con), NULL)) - expect_true(resp_stream_is_complete(con)) -}) - test_that("can't read from a closed connection", { resp <- request_test("/stream-bytes/1024") |> req_perform_connection() close(resp) @@ -64,272 +47,170 @@ test_that("can't read from a closed connection", { expect_no_error(close(resp)) }) -test_that("can join lines across multiple reads", { - sync <- sync_req("join") - req <- local_app_request(function(req, res) { - sync <- req$app$locals$sync_rep("join") - - res$send_chunk("This is a ") - sync(res$send_chunk("complete sentence.\n")) - }) +test_that("streaming functions require a streaming response", { + expect_snapshot(resp_stream_raw(response()), error = TRUE) +}) - resp1 <- req_perform_connection(req, blocking = FALSE) - withr::defer(close(resp1)) - wait_for_http_data(resp1) +test_that("resp_stream_raw() validates kb", { + resp <- local_streaming_response(charToRaw("abc")) - out <- resp_stream_lines(resp1) - expect_equal(out, character()) - expect_equal(resp1$cache$push_back, charToRaw("This is a ")) + expect_length(resp_stream_raw(resp, kb = 1 / 1024), 1) + expect_snapshot(resp_stream_raw(resp, kb = -1), error = TRUE) + expect_snapshot(resp_stream_raw(resp, kb = Inf), error = TRUE) +}) - out <- resp_stream_lines(resp1) - expect_equal(out, character()) +test_that("resp_stream_is_complete() requires an open streaming response", { + expect_snapshot(resp_stream_is_complete(response()), error = TRUE) - sync(resp1) - out <- resp_stream_lines(resp1) - expect_equal(out, "This is a complete sentence.") + resp <- local_streaming_response(charToRaw("abc")) + close(resp) + expect_snapshot(resp_stream_is_complete(resp), error = TRUE) }) -test_that("handles line endings of multiple kinds", { - sync <- sync_req("endings") - req <- local_app_request(function(req, res) { - sync <- req$app$locals$sync_rep("endings") - - res$set_header("Content-Type", "text/plain; charset=Shift_JIS") - res$send_chunk(as.raw(c(0x82, 0xA0, 0x0A))) - sync(res$send_chunk("crlf\r\n")) - sync(res$send_chunk("lf\n")) - sync(res$send_chunk("cr\r")) - sync(res$send_chunk("half line/")) - sync(res$send_chunk("other half\n")) - sync(res$send_chunk("broken crlf\r")) - sync(res$send_chunk("\nanother line\n")) - sync(res$send_chunk("eof without line ending")) - }) +test_that("streaming responses use only one reader", { + resp <- local_streaming_response(charToRaw("a\n")) - resp1 <- req_perform_connection(req, blocking = FALSE) - withr::defer(close(resp1)) - wait_for_http_data(resp1) + expect_equal(resp_stream_lines(resp), "a") + expect_snapshot(resp_stream_raw(resp), error = TRUE) +}) - expected_values <- list( - "\u3042", - "crlf", - "lf", - "cr", - character(0), - "half line/other half", - "broken crlf", - "another line" - ) +test_that("StreamSplitter$finish() discards a trailing partial block with a warning", { + # The block-splitting and read sizing live in stream_pull() and are covered by + # its integration tests; finish() is the one behavior that stays on the class. + s <- SseSplitter$new() + # Nothing buffered: nothing to flush, no warning. + expect_equal(s$finish(raw()), list()) + # A trailing partial block is dropped with a warning. + expect_snapshot(out <- s$finish(charToRaw("b"))) + expect_equal(out, list()) +}) - for (expected in expected_values) { - rlang::inject(expect_equal(resp_stream_lines(resp1), !!expected)) - sync(resp1) - } - wait_for_complete(resp1) - expect_warning(out <- resp_stream_lines(resp1), "incomplete final line") - expect_equal(out, "eof without line ending") - expect_equal(resp_stream_lines(resp1), character(0)) +# stream_pull() drives every format (lines, sse, aws); these tests exercise its +# format-independent behavior through resp_stream_lines() as a convenient +# vehicle. Format-specific splitting and parsing are tested in the per-format +# files. - # Same test, but now, blocking (and without sync) +test_that("stream_pull() buffers incomplete blocks across reads (non-blocking)", { + sync <- sync_req("pull") req <- local_app_request(function(req, res) { - res$set_header("Content-Type", "text/plain; charset=Shift_JIS") - res$send_chunk(as.raw(c(0x82, 0xA0, 0x0A))) - res$send_chunk("crlf\r\n") - res$send_chunk("lf\n") - res$send_chunk("cr\r") - res$send_chunk("half line/") - res$send_chunk("other half\n") - res$send_chunk("broken crlf\r") - res$send_chunk("\nanother line\n") - res$send_chunk("eof without line ending") - }) - resp2 <- req_perform_connection(req, blocking = TRUE) - withr::defer(close(resp2)) - - expected_values <- c( - "\u3042", - "crlf", - "lf", - "cr", - "half line/other half", - "broken crlf", - "another line" - ) - - for (expected in expected_values) { - rlang::inject(expect_equal(resp_stream_lines(resp2), !!expected)) - } - expect_warning(out <- resp_stream_lines(resp2), "incomplete final line") - expect_equal(out, "eof without line ending") -}) + sync <- req$app$locals$sync_rep("pull") -test_that("streams the specified number of lines", { - req <- local_app_request(function(req, res) { - res$send_chunk(paste0(letters[1:5], "\n", collapse = "")) + res$send_chunk("This is a ") + sync(res$send_chunk("complete sentence.\n")) }) - resp1 <- req_perform_connection(req, blocking = TRUE) - withr::defer(close(resp1)) - expect_equal(resp_stream_lines(resp1, 3), c("a", "b", "c")) - expect_equal(resp_stream_lines(resp1, 3), c("d", "e")) - expect_equal(resp_stream_lines(resp1, 3), character()) - - resp2 <- req_perform_connection(req, blocking = FALSE) - withr::defer(close(resp2)) - wait_for_http_data(resp2) - expect_equal(resp_stream_lines(resp2, 3), c("a", "b", "c")) - expect_equal(resp_stream_lines(resp2, 3), c("d", "e")) - expect_equal(resp_stream_lines(resp2, 3), character()) -}) - -test_that("can feed sse events one at a time", { - req <- local_app_request(function(req, res) { - for (i in 1:3) { - res$send_chunk(sprintf("data: %s\n\n", i)) - } - }) - resp <- req_perform_connection(req) + resp <- req_perform_connection(req, blocking = FALSE) withr::defer(close(resp)) + wait_for_http_data(resp) - expect_equal( - resp_stream_sse(resp), - list(type = "message", data = "1", id = "") - ) - expect_equal( - resp_stream_sse(resp), - list(type = "message", data = "2", id = "") - ) - resp_stream_sse(resp) + # An incomplete block is held in stream_buffer and nothing is served yet. + expect_equal(resp_stream_lines(resp), character()) + expect_equal(resp$cache$stream_buffer, charToRaw("This is a ")) + # Buffered bytes mean the stream isn't complete, even between blocks. + expect_false(resp_stream_is_complete(resp)) + + expect_equal(resp_stream_lines(resp), character()) - expect_equal(resp_stream_sse(resp), NULL) + sync(resp) + expect_equal(resp_stream_lines(resp), "This is a complete sentence.") }) -test_that("ignores events with no data", { +test_that("stream_pull() serves queued blocks from a single read", { req <- local_app_request(function(req, res) { - res$send_chunk(": comment\n\n") - res$send_chunk("data: 1\n\n") + res$send_chunk(paste0(letters[1:5], "\n", collapse = "")) }) - resp <- req_perform_connection(req) + + resp <- req_perform_connection(req, blocking = TRUE) withr::defer(close(resp)) - expect_equal( - resp_stream_sse(resp), - list(type = "message", data = "1", id = "") - ) + # All five blocks are split from one read; `n` limits how many are served and + # the rest stay queued, so is_complete() is FALSE while the queue is non-empty. + expect_equal(resp_stream_lines(resp, 3), c("a", "b", "c")) + expect_false(resp_stream_is_complete(resp)) + expect_equal(resp_stream_lines(resp, 3), c("d", "e")) + expect_equal(resp_stream_lines(resp, 3), character()) + expect_true(resp_stream_is_complete(resp)) }) -test_that("can join sse events across multiple reads", { - sync <- sync_req("sse") +test_that("stream_pull() serves every block when n is Inf", { req <- local_app_request(function(req, res) { - sync <- req$app$locals$sync_rep("sse") - - res$send_chunk("data: 1\n") - sync(res$send_chunk("data")) - res$send_chunk(": 2\n") - sync(res$send_chunk("\ndata: 3\n\n")) + res$send_chunk(paste0(letters[1:5], "\n", collapse = "")) }) - # Non-blocking returns NULL until data is ready - resp1 <- req_perform_connection(req, blocking = FALSE) - withr::defer(close(resp1)) - wait_for_http_data(resp1) - - out <- resp_stream_sse(resp1) - expect_equal(out, NULL) - expect_equal(resp1$cache$push_back, charToRaw("data: 1\n")) - - sync(resp1) - out <- resp_stream_sse(resp1) - expect_equal(out, NULL) - - 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")) - - out <- resp_stream_sse(resp1) - expect_equal(out, list(type = "message", data = "3", id = "")) - - # # Blocking waits for a complete event - req <- local_app_request(function(req, res) { - res$send_chunk("data: 1\n") - res$send_chunk("data") - res$send_chunk(": 2\n") - res$send_chunk("\ndata: 3\n\n") - }) - resp2 <- req_perform_connection(req) - withr::defer(close(resp2)) + resp <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp)) - out <- resp_stream_sse(resp2) - expect_equal(out, list(type = "message", data = "1\n2", id = "")) + # n = Inf keeps serving past the queue and reads on until the stream ends. + expect_equal(resp_stream_lines(resp, Inf), letters[1:5]) + expect_equal(resp_stream_lines(resp, Inf), character()) + expect_true(resp_stream_is_complete(resp)) }) -test_that("sse always interprets data as UTF-8", { +test_that("stream_pull() enforces max_size (blocking and non-blocking)", { req <- local_app_request(function(req, res) { - res$send_chunk("data: \xE3\x81\x82\r\n\r\n") + res$send_chunk(paste(rep_len("0", 1000), collapse = "")) }) - withr::local_locale(LC_CTYPE = "C") - # Non-blocking returns NULL until data is ready resp1 <- req_perform_connection(req, blocking = FALSE) withr::defer(close(resp1)) wait_for_http_data(resp1) + expect_error( + resp_stream_lines(resp1, max_size = 999), + class = "httr2_streaming_error" + ) - out <- resp_stream_sse(resp1) - - s <- "\xE3\x81\x82" - Encoding(s) <- "UTF-8" - expect_equal(out, list(type = "message", data = s, id = "")) - expect_equal(Encoding(out$data), "UTF-8") - expect_equal(resp1$cache$push_back, raw()) + resp2 <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp2)) + expect_error( + resp_stream_lines(resp2, max_size = 999), + class = "httr2_streaming_error" + ) }) -test_that("streaming size limits enforced", { - req <- local_app_request(function(req, res) { - data_size <- 1000 - data <- paste(rep_len("0", data_size), collapse = "") - res$send_chunk(data) - }) +test_that("stream_pull() keeps size errors reproducible on retry", { + resp <- local_streaming_response(c(rep(as.raw(0x61), 11), as.raw(0x0A))) - resp1 <- req_perform_connection(req, blocking = FALSE) - withr::defer(close(resp1)) - wait_for_http_data(resp1) expect_error( - out <- resp_stream_sse(resp1, max_size = 999), + resp_stream_lines(resp, max_size = 10), + class = "httr2_streaming_error" + ) + expect_error( + resp_stream_lines(resp, max_size = 10), class = "httr2_streaming_error" ) +}) + +test_that("stream_pull() re-serves queued blocks after a failed call", { + # "a" splits off cleanly, but the long second line trips max_size, so the + # call errors *after* queueing "a". The queue is checkpointed, so retrying + # (here with a larger limit) still yields "a" rather than dropping it. + resp <- local_streaming_response(c(charToRaw("a\n"), rep(as.raw(0x30), 13))) - resp2 <- req_perform_connection(req, blocking = TRUE) - withr::defer(close(resp2)) expect_error( - out <- resp_stream_sse(resp2, max_size = 999), + resp_stream_lines(resp, lines = 2, max_size = 10), class = "httr2_streaming_error" ) + expect_equal( + resp_stream_lines(resp, lines = 2, max_size = 100), + c("a", strrep("0", 13)) + ) }) -test_that("verbosity = 2 streams request bodies", { +test_that("stream_pull() flushes a trailing block at end of stream", { req <- local_app_request(function(req, res) { - res$send_chunk("line 1\n") - res$send_chunk("line 2\n") + # "b" has no trailing line ending, so it can't be served until EOF. + res$send_chunk("a\nb") }) + resp <- req_perform_connection(req, blocking = TRUE) + withr::defer(close(resp)) - stream_all <- function(req, fun, ...) { - con <- req_perform_connection(req, blocking = TRUE, verbosity = 2) - withr::defer(close(con)) - while (!resp_stream_is_complete(con)) { - fun(con, ...) - } - } - expect_snapshot( - { - stream_all(req, resp_stream_lines, 1) - stream_all(req, resp_stream_raw, 5 / 1024) - }, - transform = function(lines) lines[!grepl("^(<-|->)", lines)] - ) + expect_equal(resp_stream_lines(resp, 1), "a") + # At end of stream the splitter flushes its buffered remainder as a block. + expect_equal(resp_stream_lines(resp, 1), "b") + expect_equal(resp_stream_lines(resp, 1), character()) }) -test_that("verbosity = 3 shows buffer info", { +test_that("verbosity = 3 logs the buffered chunk", { req <- local_app_request(function(req, res) { res$send_chunk("line 1\n") res$send_chunk("line 2\n") @@ -349,74 +230,24 @@ test_that("verbosity = 3 shows buffer info", { ) }) -test_that("verbosity = 3 shows raw sse events", { +test_that("verbosity = 2 streams request bodies", { req <- local_app_request(function(req, res) { - res$send_chunk(": comment\n\n") - res$send_chunk("data: 1\n\n") + res$send_chunk("line 1\n") + res$send_chunk("line 2\n") }) - expect_output(resp <- req_perform_connection(req, verbosity = 3)) - withr::defer(close(resp)) - expect_snapshot( - . <- resp_stream_sse(resp), - transform = transform_verbose_response - ) -}) - -test_that("has a working find_event_boundary", { - boundary_test <- function(x, matched, remaining) { - buffer <- charToRaw(x) - split_at <- find_event_boundary(buffer) - result <- if (is.null(split_at)) { - NULL - } else { - split_buffer(buffer, split_at) + stream_all <- function(req, fun, ...) { + con <- req_perform_connection(req, blocking = TRUE, verbosity = 2) + withr::defer(close(con)) + while (!resp_stream_is_complete(con)) { + fun(con, ...) } - expect_identical( - result, - list(matched = charToRaw(matched), remaining = charToRaw(remaining)) - ) } - - # Basic matches - boundary_test("\r\r", matched = "\r\r", remaining = "") - boundary_test("\n\n", matched = "\n\n", remaining = "") - boundary_test("\r\n\r\n", matched = "\r\n\r\n", remaining = "") - boundary_test("a\r\r", matched = "a\r\r", remaining = "") - boundary_test("a\n\n", matched = "a\n\n", remaining = "") - boundary_test("a\r\n\r\n", matched = "a\r\n\r\n", remaining = "") - boundary_test("\r\ra", matched = "\r\r", remaining = "a") - boundary_test("\n\na", matched = "\n\n", remaining = "a") - boundary_test("\r\n\r\na", matched = "\r\n\r\n", remaining = "a") - - # Matches the first boundary found - boundary_test("\r\r\r", matched = "\r\r", remaining = "\r") - boundary_test("\r\r\r\r", matched = "\r\r", remaining = "\r\r") - boundary_test("\n\n\r\r", matched = "\n\n", remaining = "\r\r") - boundary_test("\r\r\n\n", matched = "\r\r", remaining = "\n\n") - - # Non-matches - expect_null(find_event_boundary(charToRaw("\n\r\n\r"))) - expect_null(find_event_boundary(charToRaw("hello\ngoodbye\n"))) - expect_null(find_event_boundary(charToRaw(""))) - expect_null(find_event_boundary(charToRaw("1"))) - expect_null(find_event_boundary(charToRaw("12"))) - expect_null(find_event_boundary(charToRaw("\r\n\r"))) -}) - -# parse_event ---------------------------------------------------------------- - -test_that("event with no data returns NULL", { - expect_null(parse_event("")) - expect_null(parse_event(":comment")) - expect_null(parse_event("id: 1")) - - expect_equal(parse_event("data: ")$data, "") - expect_equal(parse_event("data")$data, "") -}) - -test_that("examples from spec work", { - event <- parse_event("data: YHOO\ndata: +2\ndata: 10") - expect_equal(event$type, "message") - expect_equal(event$data, "YHOO\n+2\n10") + expect_snapshot( + { + stream_all(req, resp_stream_lines, 1) + stream_all(req, resp_stream_raw, 5 / 1024) + }, + transform = function(lines) lines[!grepl("^(<-|->)", lines)] + ) })