Refine resp_stream_sse()

NEWS.md

@@ -1,7 +1,7 @@
 # httr2 (development version)
 
-* `aws_v4_signature()` now works if url contains query parameters (@
-jeffreyzuber, #645).
+* `resp_stream_sse()` now automatically retrieves the next event if the current event contains no data. The data is now returned as a single string (#650).
+* `aws_v4_signature()` now works if url contains query parameters (@jeffreyzuber, #645).
 
 # httr2 1.1.0
 

R/req-perform-connection.R

@@ -23,8 +23,8 @@
 #'   * `0`: no output
 #'   * `1`: show headers
 #'   * `2`: show headers and bodies as they're streamed
-#'   * `3`: show headers, bodies, curl status messages, and stream buffer
-#'     management
+#'   * `3`: show headers, bodies, curl status messages, raw SSEs, and stream
+#'     buffer management
 #'
 #'   Use [with_verbosity()] to control the verbosity of requests that
 #'   you can't affect directly.

R/resp-stream.R

@@ -14,7 +14,9 @@
 #' @returns
 #' * `resp_stream_raw()`: a raw vector.
 #' * `resp_stream_lines()`: a character vector.
-#' * `resp_stream_sse()`: a list with components `type`, `data`, and `id`
+#' * `resp_stream_sse()`: a list with components `type`, `data`, and `id`.
+#'   `type`, `data`, and `id` are always strings; `data` and `id` may be empty
+#'   strings.
 #' * `resp_stream_aws()`: a list with components `headers` and `body`.
 #'   `body` will be automatically parsed if the event contents a `:content-type`
 #'   header with `application/json`.
@@ -88,7 +90,6 @@ resp_stream_lines <- function(resp, lines = 1, max_size = Inf, warn = TRUE) {
 
   if (resp_stream_show_body(resp)) {
     log_stream(lines_read)
-    cli::cat_line()
   }
 
   lines_read
@@ -101,12 +102,26 @@ resp_stream_lines <- function(resp, lines = 1, max_size = Inf, warn = TRUE) {
 #' @rdname resp_stream_raw
 #' @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)) {
-    return()
+
+  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
   }
 
-  event <- parse_event(event_bytes)
   if (resp_stream_show_body(resp)) {
     for (key in names(event)) {
       log_stream(cli::style_bold(key), ": ", pretty_json(event[[key]]))
@@ -263,8 +278,9 @@ resp_boundary_pushback <- function(resp, max_size, boundary_func, include_traile
   resp$cache$push_back <- raw()
 
   if (resp_stream_show_buffer(resp)) {
+    log_stream(cli::rule("Buffer"), prefix = " * ")
     print_buffer <- function(buf, label) {
-      cli::cat_line(" * ", label, ": ", paste(as.character(buf), collapse = " "))
+      log_stream(label, ": ", paste(as.character(buf), collapse = " "), prefix = " * ")
     }
   } else {
     print_buffer <- function(buf, label) {}
@@ -329,31 +345,84 @@ resp_boundary_pushback <- function(resp, max_size, boundary_func, include_traile
   }
 }
 
+# https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation
 parse_event <- function(event_data) {
-  # always treat event_data as UTF-8, it's in the spec
-  str_data <- rawToChar(event_data)
-  Encoding(str_data) <- "UTF-8"
 
-  # The spec says \r\n, \r, and \n are all valid separators
+  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)))
 
-  remove_dupes <- duplicated(keys, fromLast = TRUE) & keys != "data"
-  keys <- keys[!remove_dupes]
-  values <- values[!remove_dupes]
+  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
+    }
+  }
 
-  event_type <- values[keys == "event"]
-  data <- values[keys == "data"]
-  id <- values[keys == "id"]
+  # 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 = event_type,
+    type = type,
     data = data,
-    id = id
+    id = last_id
   )
 }
 

tests/testthat/_snaps/resp-stream.md

@@ -20,9 +20,7 @@
       stream_all(req, resp_stream_lines, 1)
     Output
       << line 1
-
       << line 2
-
     Code
       stream_all(req, resp_stream_raw, 5 / 1024)
     Output
@@ -40,17 +38,46 @@
         resp_stream_lines(con, 1)
       }
     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
+
+    Code
+      . <- resp_stream_sse(resp)
+    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: 
 
 

tests/testthat/test-resp-stream.R

@@ -42,7 +42,7 @@ test_that("can determine if incomplete data is complete", {
   })
 
   con <- req %>% req_perform_connection(blocking = TRUE)
-  expect_equal(resp_stream_sse(con, 10), list(type = "message", data = "1", id = character()))
+  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))
   close(con)
@@ -188,17 +188,31 @@ test_that("can feed sse events one at a time", {
 
   expect_equal(
     resp_stream_sse(resp),
-    list(type = "message", data = "1", id = character())
+    list(type = "message", data = "1", id = "")
   )
   expect_equal(
     resp_stream_sse(resp),
-    list(type = "message", data = "2", id = character())
+    list(type = "message", data = "2", id = "")
   )
   resp_stream_sse(resp)
 
   expect_equal(resp_stream_sse(resp), NULL)
 })
 
+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("can join sse events across multiple reads", {
   req <- local_app_request(function(req, res) {
     res$send_chunk("data: 1\n")
@@ -221,17 +235,17 @@ test_that("can join sse events across multiple reads", {
     Sys.sleep(0.1)
     out <- resp_stream_sse(resp1)
   }
-  expect_equal(out, list(type = "message", data = c("1", "2"), id = character()))
+  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 = character()))
+  expect_equal(out, list(type = "message", data = "3", id = ""))
 
   # Blocking waits for a complete event
   resp2 <- req_perform_connection(req)
   withr::defer(close(resp2))
 
   out <- resp_stream_sse(resp2)
-  expect_equal(out, list(type = "message", data = c("1", "2"), id = character()))
+  expect_equal(out, list(type = "message", data = "1\n2", id = ""))
 })
 
 test_that("sse always interprets data as UTF-8", {
@@ -252,7 +266,7 @@ test_that("sse always interprets data as UTF-8", {
 
     s <- "\xE3\x81\x82"
     Encoding(s) <- "UTF-8"
-    expect_equal(out, list(type = "message", data = s, id = character()))
+    expect_equal(out, list(type = "message", data = s, id = ""))
     expect_equal(Encoding(out$data), "UTF-8")
     expect_equal(resp1$cache$push_back, raw())
   })
@@ -327,6 +341,16 @@ test_that("verbosity = 3 shows buffer info", {
   )
 })
 
+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")
+  })
+
+  resp <- req_perform_connection(req, verbosity = 3)
+  withr::defer(close(resp))
+  expect_snapshot(. <- resp_stream_sse(resp))
+})
 
 test_that("has a working find_event_boundary", {
   boundary_test <- function(x, matched, remaining) {
@@ -368,3 +392,20 @@ test_that("has a working find_event_boundary", {
   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")
+})