Add user interrupt check while waiting for query results to be ready.

DESCRIPTION

@@ -1,11 +1,12 @@
 Package: RPostgres
 Title: 'Rcpp' Interface to 'PostgreSQL'
-Version: 1.1.1
-Date: 2018-05-05
+Version: 1.1.2
+Date: 2018-08-25
 Authors@R: c(
     person("Hadley", "Wickham", role = "aut"),
     person("Jeroen", "Ooms", role = "aut"),
     person("Kirill", "M\u00fcller", role = c("aut", "cre"), email = "krlmlr+r@mailbox.org", comment = c(ORCID = "0000-0002-1416-3412")),
+    person("Mateusz", "\u017b\u00f3\u0142tak", role = c("ctb")),
     person("RStudio", role = "cph"),
     person("R Consortium", role = "fnd"),
     person("Tomoaki", "Nishiyama", role = "ctb",

NEWS.md

@@ -1,3 +1,7 @@
+# RPostgres 1.1.2 (2018-08-25)
+
+- Add user interrupt check while waiting for query results to be ready. Allows safe interruption of long-running queries. 
+
 # RPostgres 1.1.1 (2018-05-05)
 
 - Add support for `bigint` argument to `dbConnect()`, supported values are `"integer64"`, `"integer"`, `"numeric"` and `"character"`. Large integers are returned as values of that type (r-dbi/DBItest#133).

src/PqResultImpl.cpp

@@ -12,6 +12,7 @@ PqResultImpl::PqResultImpl(DbResult* pRes, PGconn* pConn, const std::string& sql
   cache(pSpec_),
   complete_(false),
   ready_(false),
+  data_ready_(false),
   nrows_(0),
   rows_affected_(0),
   group_(0),
@@ -329,6 +330,7 @@ bool PqResultImpl::bind_row() {
     PQsendQueryPrepared(pConn_, "", cache.nparams_, &c_params[0],
                         &lengths[0], &formats[0], 0) :
     PQsendQueryPrepared(pConn_, "", 0, NULL, NULL, NULL, 0);
+  data_ready_ = false;
 
   if (!success)
     conn_stop("Failed to send query");
@@ -379,6 +381,35 @@ bool PqResultImpl::step_run() {
   LOG_VERBOSE;
 
   if (pRes_) PQclear(pRes_);
+
+  // Check user interrupts while waiting for the data to be ready
+  if (!data_ready_) {
+    int socket, ret;
+    fd_set input;
+    timeval timeout = {0, 0};
+
+    socket = PQsocket(pConn_);
+    if (socket < 0) {
+      stop("Failed to get connection socket");
+    }
+    FD_ZERO(&input);
+    FD_SET(socket, &input);
+
+    do {
+      timeout.tv_sec = 1;
+      ret = select(socket + 1, &input, NULL, NULL, &timeout);
+      if (ret == 0) {
+        checkUserInterrupt();
+      } else if(ret < 0) {
+        stop("select() on the connection failed");
+      }
+      if (!PQconsumeInput(pConn_)) {
+        stop("Failed to consume input from the server");
+      }
+    } while (PQisBusy(pConn_));
+    data_ready_ = true;
+  }
+
   pRes_ = PQgetResult(pConn_);
 
   // We're done, but we need to call PQgetResult until it returns NULL

src/PqResultImpl.h

@@ -37,6 +37,7 @@ class PqResultImpl : boost::noncopyable, public PqResultSource {
   // State
   bool complete_;
   bool ready_;
+  bool data_ready_;
   int nrows_;
   int rows_affected_;
   List params_;