use active bindings for more intuitive Aio UI

Author: shikokuchuo

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: nanonext
 Type: Package
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 0.2.0.9002
+Version: 0.2.0.9003
 Description: R binding for NNG (Nanomsg Next Gen), a successor to ZeroMQ. NNG is 
     a socket library providing high-performance scalability protocols,
     implementing a cross-platform standard for messaging and communications.

NAMESPACE

@@ -15,6 +15,7 @@ S3method(print,nanoObject)
 S3method(print,nanoSocket)
 S3method(print,recvAio)
 S3method(print,sendAio)
+S3method(print,unresolvedValue)
 S3method(setopt,nanoContext)
 S3method(setopt,nanoDialer)
 S3method(setopt,nanoListener)
@@ -26,6 +27,7 @@ export(call_aio)
 export(context)
 export(dial)
 export(is_nul_byte)
+export(is_resolved)
 export(listen)
 export(nano)
 export(ncurl)

NEWS.md

@@ -1,15 +1,16 @@
-# nanonext 0.2.0.9002 (development)
+# nanonext 0.2.0.9003 (development)
 
 #### New Features
 
-* `call_aio()` gains the argument 'block', offering a non-blocking method of calling an AIO, without waiting for completion if it is yet to resolve.
-* `survey_time()` added as a convenience function for surveyor/respondent patterns.
+* Aio fields `$result` for a 'sendAio', `$raw` and `$data` for a 'recvAio' may be queried directly, returning their values or else an NA 'unresolved value' if the Aio operation is yet to complete.
+* `is_resolved()` added as an auxiliary function to query whether an Aio has resolved in a non-blocking fashion.
 * `is_nul_byte()` added as a helper function for request/reply setups.
+* `survey_time()` added as a convenience function for surveyor/respondent patterns.
 * `ncurl()` adds a '...' argument. Support for HTTP methods other than GET.
 
 #### Updates
 
-* 'quietly' argument added to functions that create or destroy objects. Set to FALSE to enable printing of informational messages for logging purposes.
+* Argument 'quietly' added to functions that create or destroy objects. Set to FALSE to enable printing of informational messages for logging purposes.
 * Common format for NNG errors now starts with a timestamp for easier logging.
 * `listen()` and `dial()` now return (invisible) zero rather than NULL upon success to better align with similar functions.
 * Allows setting the environment variable 'NANONEXT_ARM' prior to package installation

R/aio.R

@@ -2,15 +2,12 @@
 
 #' Call the Result of an Asynchronous AIO Operation
 #'
-#' Retrieve the result of an asynchronous AIO operation. Optionally wait for the
-#'     AIO operation to complete if this is still in progress.
+#' Retrieve the result of an asynchronous AIO operation, waiting for the AIO
+#'     operation to complete if still in progress.
 #'
 #' @param aio An Aio (object of class 'sendAio' or 'recvAio').
-#' @param block [default TRUE] whether to wait for completion of the AIO
-#'     operation (blocking) or return immediately.
 #'
-#' @return The passed Aio object (invisibly), or NULL if non-blocking and the
-#'     Aio has yet to resolve.
+#' @return The passed Aio object (invisibly).
 #'
 #' @details For a 'recvAio', the received raw vector will be attached in \code{$raw}
 #'     (unless 'keep.raw' was set to FALSE when receiving), and the converted R
@@ -23,23 +20,21 @@
 #'     \code{call_aio(x)$result}.
 #'
 #'     For a 'recvAio', in case of an error in unserialisation or data conversion,
-#'     the received raw vector will always be saved in \code{$raw} to allow the
+#'     the received raw vector will be stored in \code{$data} to allow for the
 #'     data to be recovered.
 #'
 #'     Once the result has been successfully retrieved, the Aio is deallocated
 #'     and only the result is stored in the Aio object.
 #'
 #' @section Non-blocking:
 #'
-#'     To query whether Aio \code{x} has resolved, test if \code{call_aio(x, block = FALSE)}
-#'     returns NULL. When the Aio resolves, the Aio itself will be returned
-#'     (invisibly) instead of NULL. The data may then be extracted from the Aio
-#'     using \code{$result}, \code{$raw} or \code{$data} as the case may be.
+#'     To query the value of an Aio without potentially waiting for the Aio
+#'     operation to complete, call the values directly at \code{$result} for a 'sendAio', and
+#'     \code{$raw} or \code{$data} for a 'recvAio'.
 #'
-#'     It is not advisable to use, for example, \code{call_aio(x, block = FALSE)$data}.
-#'     This is as \code{call_aio()} will return NULL if the Aio is unresolved and
-#'     \code{NULL$data} is also \code{NULL}, hence it would be impossible to
-#'     distinguish between an unresolved Aio and a NULL return value in this case.
+#'     If the Aio operation is yet to complete, the result will be an
+#'     'unresolved value', which is a logical NA. Once complete, the resolved
+#'     value will be returned instead.
 #'
 #' @examples
 #' s1 <- socket("pair", listen = "inproc://nanonext")
@@ -48,40 +43,40 @@
 #' res <- send_aio(s1, data.frame(a = 1, b = 2), timeout = 100)
 #' res
 #' call_aio(res)
-#' res
 #' res$result
 #'
 #' res <- recv_aio(s2, timeout = 100)
 #' res
 #' call_aio(res)$data
-#' res
 #'
 #' close(s1)
 #' close(s2)
 #'
 #' @export
 #'
-call_aio <- function(aio, block = TRUE) {
+call_aio <- function(aio) {
 
   is.null(.subset2(aio, "aio")) && return(invisible(aio))
 
   if (inherits(aio, "recvAio")) {
 
-    if (missing(block) || isTRUE(block)) {
-      res <- .Call(rnng_aio_wait_get_msg, .subset2(aio, "aio"))
-    } else {
-      res <- .Call(rnng_aio_get_msg, .subset2(aio, "aio"))
-      missing(res) && return()
-    }
+    res <- .Call(rnng_aio_wait_get_msg, .subset2(aio, "aio"))
     mode <- .subset2(aio, "callparams")[[1L]]
     keep.raw <- .subset2(aio, "callparams")[[2L]]
-    if (keep.raw) aio[["raw"]] <- res
     is.integer(res) && {
+      if (keep.raw) {
+        rm("raw", envir = aio)
+        `[[<-`(aio, "raw", res)
+      }
+      rm("data", envir = aio)
+      `[[<-`(aio, "data", res)
       message(Sys.time(), " [ ", res, " ] ", nng_error(res))
       return(invisible(aio))
     }
     on.exit(expr = {
-      aio[["raw"]] <- res
+      if (keep.raw) rm("raw", envir = aio)
+      rm("data", envir = aio)
+      `[[<-`(aio, "data", res)
       rm("aio", envir = aio)
       rm("callparams", envir = aio)
       return(invisible(aio))
@@ -91,22 +86,23 @@ call_aio <- function(aio, block = TRUE) {
                    character = (r <- readBin(con = res, what = mode, n = length(res)))[r != ""],
                    raw = res,
                    readBin(con = res, what = mode, n = length(res)))
-    aio[["data"]] <- data
     on.exit()
+    if (keep.raw) {
+      rm("raw", envir = aio)
+      `[[<-`(aio, "raw", res)
+    }
+    rm("data", envir = aio)
+    `[[<-`(aio, "data", data)
     rm("aio", envir = aio)
     rm("callparams", envir = aio)
 
   } else if (inherits(aio, "sendAio")) {
 
-    if (missing(block) || isTRUE(block)) {
-      res <- .Call(rnng_aio_wait_result, .subset2(aio, "aio"))
-    } else {
-      res <- .Call(rnng_aio_result, .subset2(aio, "aio"))
-      missing(res) && return()
-    }
-    aio[["result"]] <- res
+    res <- .Call(rnng_aio_wait_result, .subset2(aio, "aio"))
+    rm("result", envir = aio)
+    `[[<-`(aio, "result", res)
     rm("aio", envir = aio)
-    if (res) message(Sys.time(), " | ", res, " : ", nng_error(res))
+    if (res) message(Sys.time(), " [ ", res, " ] ", nng_error(res))
 
   }
 
@@ -135,3 +131,40 @@ stop_aio <- function(aio) {
 
 }
 
+#' Is Resolved (Asynchronous AIO Operation)
+#'
+#' Query whether an Aio has resolved. This function is non-blocking unlike
+#'     \code{\link{call_aio}} which waits for completion.
+#'
+#' @param aio An Aio (object of class 'sendAio' or 'recvAio').
+#'
+#' @return Logical TRUE or FALSE. NA if 'aio' is not a 'sendAio' or 'recvAio'.
+#'
+#' @details Querying resolution will potentially cause the state of the Aio to
+#'     update.
+#'
+#' @examples
+#' s1 <- socket("pair", listen = "inproc://nanonext")
+#' aio <- send_aio(s1, "test", timeout = 100)
+#' is_resolved(aio)
+#'
+#' s2 <- socket("pair", dial = "inproc://nanonext")
+#' is_resolved(aio)
+#'
+#' close(s1)
+#' close(s2)
+#'
+#' @export
+#'
+is_resolved <- function(aio) {
+
+  if (inherits(aio, "recvAio")) {
+    !inherits(aio$data, "unresolvedValue")
+  } else if (inherits(aio, "sendAio")) {
+    !inherits(aio$result, "unresolvedValue")
+  } else {
+    NA
+  }
+
+}
+

R/context.R

@@ -312,6 +312,7 @@ request <- function(context,
 
   send_mode <- match.arg(send_mode)
   recv_mode <- match.arg(recv_mode)
+  keep.raw <- missing(keep.raw) || isTRUE(keep.raw)
   if (missing(timeout)) timeout <- -2L
   force(data)
   data <- switch(send_mode,
@@ -322,15 +323,70 @@ request <- function(context,
     message(Sys.time(), " [ ", res, " ] ", nng_error(res))
     return(invisible(res))
   }
-  res <- .Call(rnng_recv_aio, context, timeout)
-  is.integer(res) && {
-    message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-    return(invisible(res))
+
+  aio <- .Call(rnng_recv_aio, context, timeout)
+  is.integer(aio) && {
+    message(Sys.time(), " [ ", aio, " ] ", nng_error(aio))
+    return(invisible(aio))
   }
   env <- `class<-`(new.env(), "recvAio")
-  env[["aio"]] <- res
-  env[["callparams"]] <- list(recv_mode, missing(keep.raw) || isTRUE(keep.raw))
-  env
+  `[[<-`(env, "callparams", list(recv_mode, keep.raw))
+  data <- raw <- resolv <- NULL
+  if (keep.raw) {
+    makeActiveBinding(sym = "raw", fun = function(x) {
+      if (is.null(resolv)) {
+        res <- .Call(rnng_aio_get_msg, aio)
+        missing(res) && return(.Call(rnng_aio_unresolv))
+        is.integer(res) && {
+          res <<- raw <<- resolv <<- res
+          message(Sys.time(), " [ ", res, " ] ", nng_error(res))
+          return(invisible(res))
+        }
+        on.exit(expr = {
+          raw <<- res
+          resolv <<- 0L
+          return(res)
+        })
+        data <- switch(recv_mode,
+                       serial = unserialize(connection = res),
+                       character = (r <- readBin(con = res, what = recv_mode, n = length(res)))[r != ""],
+                       raw = res,
+                       readBin(con = res, what = recv_mode, n = length(res)))
+        on.exit()
+        raw <<- res
+        data <<- data
+        resolv <<- 0L
+      }
+      raw
+    }, env = env)
+  }
+  makeActiveBinding(sym = "data", fun = function(x) {
+    if (is.null(resolv)) {
+      res <- .Call(rnng_aio_get_msg, aio)
+      missing(res) && return(.Call(rnng_aio_unresolv))
+      is.integer(res) && {
+        res <<- raw <<- resolv <<- res
+        message(Sys.time(), " [ ", res, " ] ", nng_error(res))
+        return(invisible(res))
+      }
+      on.exit(expr = {
+        data <<- res
+        resolv <<- 0L
+        return(res)
+      })
+      data <- switch(recv_mode,
+                     serial = unserialize(connection = res),
+                     character = (r <- readBin(con = res, what = recv_mode, n = length(res)))[r != ""],
+                     raw = res,
+                     readBin(con = res, what = recv_mode, n = length(res)))
+      on.exit()
+      if (keep.raw) raw <<- res
+      data <<- data
+      resolv <<- 0L
+    }
+    data
+  }, env = env)
+  `[[<-`(env, "aio", aio)
 
 }
 

R/docs.R

@@ -349,10 +349,10 @@ NULL
 #'
 #'     \item{'socket-name' [type 'string']}
 #'
-#'       {This the socket name. By default this is a string corresponding to the
-#'       value of the socket. The string must fit within 64-bytes, including the
-#'       terminating NUL byte. The value is intended for application use, and is
-#'       not used for anything in the library itself.}
+#'       {This is the socket name. By default this is a string corresponding to
+#'       the value of the socket. The string must fit within 64-bytes, including
+#'       the terminating NUL byte. The value is intended for application use,
+#'       and is not used for anything in the library itself.}
 #'
 #'     }
 #'

R/nano.R

@@ -217,15 +217,7 @@ print.nanoListener <- function(x, ...) {
 #'
 print.recvAio <- function(x, ...) {
 
-  cat("< recvAio >\n")
-  if (length(.subset2(x, "aio"))) {
-    cat(" ~ use call_aio() to retrieve message\n")
-  } else {
-    if (length(.subset2(x, "raw")))
-      cat(" - $raw for raw message\n")
-    if (length(.subset2(x, "data")))
-      cat(" - $data for message data\n")
-  }
+  cat("< recvAio >\n - $raw for raw message\n - $data for message data\n")
   invisible(x)
 
 }
@@ -234,12 +226,7 @@ print.recvAio <- function(x, ...) {
 #'
 print.sendAio <- function(x, ...) {
 
-  cat("< sendAio >\n")
-  if (length(.subset2(x, "aio"))) {
-    cat(" ~ use call_aio() to retrieve result\n")
-  } else {
-    cat(" - $result for send result\n")
-  }
+  cat("< sendAio >\n - $result for send result\n")
   invisible(x)
 
 }
@@ -276,3 +263,12 @@ print.sendAio <- function(x, ...) {
 
 }
 
+#' @export
+#'
+print.unresolvedValue <- function(x, ...) {
+
+  cat("< unresolved value >\n")
+  invisible(x)
+
+}
+

R/nanonext-package.R

@@ -122,4 +122,3 @@
 #'
 NULL
 
-utils::globalVariables("eval_mirai")