ncurlAio promises implementation

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: nanonext
 Type: Package
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 1.1.0.9003
+Version: 1.1.0.9004
 Description: R binding for NNG (Nanomsg Next Gen), a successor to ZeroMQ. NNG is
     a socket library implementing 'Scalability Protocols', a reliable,
     high-performance standard for common communications patterns including
@@ -30,6 +30,8 @@ SystemRequirements: 'libnng' >= 1.6 and 'libmbedtls' >= 2.5, or 'cmake' and 'xz'
     to compile NNG and/or Mbed TLS included in package sources
 Depends: 
     R (>= 3.5)
+Enhances:
+    promises
 Suggests: 
     knitr,
     later,

NAMESPACE

@@ -35,6 +35,8 @@ S3method(print,recvAio)
 S3method(print,sendAio)
 S3method(print,tlsConfig)
 S3method(print,unresolvedValue)
+S3method(promises::as.promise,ncurlAio)
+S3method(promises::is.promising,ncurlAio)
 S3method(start,nanoDialer)
 S3method(start,nanoListener)
 export("%~>%")

NEWS.md

@@ -1,8 +1,9 @@
-# nanonext 1.1.0.9003 (development)
+# nanonext 1.1.0.9004 (development)
 
 #### New Features
 
-* The method `x[]` for an Aio `x` is a new equivalent to `collect_aio_(x)`, which waits for and collects the data.
+* Adds 'ncurlAio' method for `promises::as.promise()` and `promises::is.promising()` to enable 'ncurlAio' promises.
+* Adds method `x[]` for an Aio `x` as a new equivalent to `collect_aio_(x)`, which waits for and collects the data.
 
 #### Updates
 

R/ncurl.R

@@ -108,6 +108,18 @@ ncurl <- function(url,
 #'     save as a file).
 #'     }
 #'
+#' @section Promises:
+#'
+#'     \sQuote{ncurlAio} may be used anywhere that accepts a \sQuote{promise}
+#'     from the \CRANpkg{promises} package through the included
+#'     \code{as.promise} method.
+#'
+#'     The promises created are completely event-driven and non-polling.
+#'
+#'     If a status code of 200 (OK) is returned then the promise is resolved
+#'     with the reponse body, otherwise it is rejected with a translation of the
+#'     status code or \sQuote{errorValue} as the case may be.
+#'
 #' @seealso \code{\link{ncurl_session}} for persistent connections.
 #' @examples
 #' nc <- ncurl_aio("https://www.r-project.org/",
@@ -118,6 +130,16 @@ ncurl <- function(url,
 #' nc$headers
 #' nc$data
 #'
+#' if (interactive() && requireNamespace("promises", quietly = TRUE)) {
+#'
+#' p <- as.promise(nc)
+#' print(p)
+#'
+#' p2 <- ncurl_aio("https://postman-echo.com/get") %...>% cat
+#' is.promise(p2)
+#'
+#' }
+#'
 #' @export
 #'
 ncurl_aio <- function(url,
@@ -189,3 +211,59 @@ transact <- function(session) .Call(rnng_ncurl_transact, session)
 #' @export
 #'
 close.ncurlSession <- function(con, ...) invisible(.Call(rnng_ncurl_session_close, con))
+
+#' Make ncurl Promise
+#'
+#' Creates a \sQuote{promise} from an \sQuote{ncurlAio} object.
+#'
+#' @param x an object of class \sQuote{ncurlAio}.
+#'
+#' @return A \sQuote{promise} object.
+#'
+#' @details This function is an S3 method for the generic \code{as.promise} for
+#'     class \sQuote{ncurlAio}.
+#'
+#'     Requires the \pkg{promises} package.
+#'
+#'     Allows an \sQuote{ncurlAio} to be used with the promise pipe
+#'     \code{\%...>\%}, which schedules a function to run upon resolution of the
+#'     Aio.
+#'
+#' @exportS3Method promises::as.promise
+#'
+as.promise.ncurlAio <- function(x) {
+
+  promise <- .subset2(x, "promise")
+
+  if (is.null(promise)) {
+
+    if (unresolved(x)) {
+      promise <- promises::then(
+        promises::promise(
+          function(resolve, reject)
+            context <- set_promise_context(x, environment())
+        ),
+        onFulfilled = function(value)
+          if (value != 200L)
+            stop(if (value < 100) nng_error(value) else status_code(value)) else
+              .subset2(x, "value")
+      )
+    } else {
+      value <- .subset2(x, "result")
+      promise <- if (value != 200L)
+        promises::promise_reject(if (value < 100) nng_error(value) else status_code(value)) else
+          promises::promise_resolve(.subset2(x, "value"))
+    }
+
+    assign("promise", promise, x)
+
+  }
+
+  promise
+
+}
+
+#' @exportS3Method promises::is.promising
+#'
+is.promising.ncurlAio <- function(x) TRUE
+

src/utils.c

@@ -25,15 +25,19 @@
 
 SEXP mk_error_ncurl(const int xc) {
 
-  const char *names[] = {"status", "headers", "data", ""};
-  SEXP out = PROTECT(Rf_mkNamed(VECSXP, names));
-  SEXP err = Rf_ScalarInteger(xc);
+  SEXP env;
+  PROTECT(env = Rf_allocSExp(ENVSXP));
+  NANO_CLASS2(env, "ncurlAio", "recvAio");
+  SEXP err = PROTECT(Rf_ScalarInteger(xc));
   Rf_classgets(err, nano_error);
-  SET_VECTOR_ELT(out, 0, err);
-  SET_VECTOR_ELT(out, 1, err);
-  SET_VECTOR_ELT(out, 2, err);
-  UNPROTECT(1);
-  return out;
+  Rf_defineVar(nano_ResultSymbol, err, env);
+  Rf_defineVar(nano_StatusSymbol, err, env);
+  Rf_defineVar(nano_ProtocolSymbol, err, env);
+  Rf_defineVar(nano_HeadersSymbol, err, env);
+  Rf_defineVar(nano_ValueSymbol, err, env);
+  Rf_defineVar(nano_DataSymbol, err, env);
+  UNPROTECT(2);
+  return env;
 
 }
 

tests/tests.R

@@ -562,6 +562,12 @@ nanotestw(dial(s, url = "tls+tcp://.", tls = tls, error = FALSE) > 0)
 nanotestw(listen(s, url = "tls+tcp://.", tls = tls, error = FALSE) > 0)
 nanotestz(close(s1))
 nanotestz(close(s))
+if (requireNamespace("promises", quietly = TRUE)) {
+  nanotestaio(n <- ncurl_aio("https://postman-echo.com/get"))
+  nanotest(tryCatch(promises::is.promise(promises::then(n, cat)), error = function(e) TRUE))
+  nanotest(promises::is.promise(promises::as.promise(call_aio(n))))
+  later::run_now()
+}
 if (Sys.info()[["sysname"]] == "Linux") {
   rm(list = ls())
   gc()