From 360eed6517b5b1483d9ad91cbcfc6d5da045d76e Mon Sep 17 00:00:00 2001
From: Hadley Wickham <h.wickham@gmail.com>
Date: Sun, 23 Jun 2024 09:08:45 +0100
Subject: [PATCH] Recover from closed sink/closed connection

I'm not 100% convinced that silently restoring the sink + connection is the correct approach, but it doesn't happen very often and we aren't going to lose much output if we immediately re-open the sink + connection.

Fixes #104 (at least as much as it can be fixed)
---
 R/watcher.R                      | 64 ++++++++++++++++++++++----------
 tests/testthat/_snaps/watcher.md | 10 +++++
 tests/testthat/test-watcher.R    | 52 ++++++++++++++++++++++++++
 3 files changed, 106 insertions(+), 20 deletions(-)
 create mode 100644 tests/testthat/_snaps/watcher.md

diff --git a/R/watcher.R b/R/watcher.R
index a5367ea..ca68edf 100644
--- a/R/watcher.R
+++ b/R/watcher.R
@@ -30,19 +30,7 @@ watchout <- function(handler = new_output_handler(),
   devn <- length(dev.list())
   dev <- dev.cur()
 
-  con <- file("", "w+b")
-  defer(frame = frame, {
-    if (!test_con(con, isOpen)) {
-      con_error('The connection has been closed')
-    }
-    sink()
-    close(con)
-  })
-  sink(con, split = debug)
-
-  # try() defaults to using stderr() so we need to explicitly override(#88)
-  old <- options(try.outFile = con)
-  defer(options(old), frame = frame)
+  sink_con <- local_persistent_sink_connection(debug, frame)
 
   capture_plot <- function(incomplete = FALSE) {
     # only record plots for our graphics device
@@ -71,7 +59,7 @@ watchout <- function(handler = new_output_handler(),
   }
 
   capture_output <- function() {
-    out <- read_con(con)
+    out <- sink_con()
     if (!is.null(out)) {
       push(out)
       handler$text(out)
@@ -100,6 +88,36 @@ watchout <- function(handler = new_output_handler(),
   )
 }
 
+# Persistent way to capture output ---------------------------------------------
+
+local_persistent_sink_connection <- function(debug = FALSE,
+                                             frame = parent.frame()) {
+  con <- file("", "w+b")
+  defer(if (isValid(con)) close(con), frame)
+
+  # try() defaults to using stderr() so we need to explicitly override(#88)
+  old <- options(try.outFile = con)
+  defer(options(old), frame)
+  
+  sink(con, split = debug)
+  sinkn <- sink.number()
+  defer(if (sink.number() >= sinkn) sink(), frame)
+
+  function() {
+    if (!isValid(con)) {
+      con <<- file("", "w+b")
+      options(try.outFile = con)
+    }
+
+    if (sink.number() < sinkn) {
+      sink(con)
+      sinkn <<- sink.number()
+    }
+
+    read_con(con)
+  }
+}
+
 read_con <- function(con, buffer = 32 * 1024) {
   bytes <- raw()
   repeat {
@@ -114,10 +132,16 @@ read_con <- function(con, buffer = 32 * 1024) {
   }
 }
 
-test_con = function(con, test) {
-  tryCatch(test(con), error = function(e) con_error(e$message))
+# 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
+  )
 }
-
-con_error = function(x) stop(
-  x, '... Please make sure not to call closeAllConnections().', call. = FALSE
-)
diff --git a/tests/testthat/_snaps/watcher.md b/tests/testthat/_snaps/watcher.md
new file mode 100644
index 0000000..d9854b4
--- /dev/null
+++ b/tests/testthat/_snaps/watcher.md
@@ -0,0 +1,10 @@
+# evaluate recovers from closed sink
+
+    Code
+      ev <- evaluate("sink()\n1")
+
+# evaluate recovers from closed connection
+
+    Code
+      ev <- evaluate("closeAllConnections()\n1")
+
diff --git a/tests/testthat/test-watcher.R b/tests/testthat/test-watcher.R
index 4d7aaed..c46c4a8 100644
--- a/tests/testthat/test-watcher.R
+++ b/tests/testthat/test-watcher.R
@@ -7,3 +7,55 @@ test_that("capture messages in try() (#88)", {
   expect_output_types(ev, c("source", "text"))
   expect_match(ev[[2]], "Obscure error")
 })
+
+test_that("code can use own sink", {
+  f <- function() {
+    con <- file("")
+    defer(close(con))
+    
+    sink(con)
+    cat("One")
+    sink()
+  }
+  ev <- evaluate("f()\n1")
+  expect_output_types(ev, c("source", "source", "text"))
+})
+
+test_that("evaluate preserves externally created sinks", {
+  sink(withr::local_tempfile())
+  defer(sink())
+  n <- sink.number()
+  
+  ev <- evaluate("1")
+  expect_output_types(ev, c("source", "text"))
+  
+  expect_equal(sink.number(), n)
+})
+
+test_that("evaluate recovers from closed sink", {
+  expect_snapshot(ev <- evaluate("sink()\n1"))
+  expect_output_types(ev, c("source", "source", "text"))
+})
+
+test_that("unbalanced sink doesn't break evaluate", {
+  path <- withr::local_tempfile()
+  ev <- evaluate("sink(path)\n1\n1")
+  expect_output_types(ev, c("source", "source", "source"))
+})
+
+test_that("evaluate recovers from closed connection", {
+  expect_snapshot(ev <- evaluate("closeAllConnections()\n1"))
+  expect_output_types(ev, c("source", "source", "text"))
+})
+
+test_that("isValid() works correctly", {
+  con1 <- file("")
+  expect_true(isValid(con1))
+  close(con1)
+  expect_false(isValid(con1))
+  
+  con2 <- file("")
+  expect_false(isValid(con1)) # isOpen would return TRUE here
+  expect_true(isValid(con2))
+  close(con2)
+})