More resilient req/rep RPC server behaviour

Author: shikokuchuo

R/context.R

@@ -165,7 +165,6 @@ ctx_recv <- function(context,
 #' @param execute a function which takes the received (converted) data as its
 #'     first argument. Can be an anonymous function of the form \code{function(x) do(x)}.
 #'     Additional arguments can also be passed in through '...'.
-#' @param ... additional arguments passed to the function specified by 'execute'.
 #' @param send_mode [default 'serial'] whether data will be sent serialized or
 #'     as a raw vector. Use 'serial' for sending and receiving within R to ensure
 #'     perfect reproducibility. Use 'raw' for sending vectors of any type (will be
@@ -179,15 +178,17 @@ ctx_recv <- function(context,
 #'     be used. Note this applies to each of the receive and send legs, hence the
 #'     total elapsed time could be up to twice this parameter plus the time to
 #'     perform 'execute' on the received data.
+#' @param ... additional arguments passed to the function specified by 'execute'.
 #'
 #' @return Invisible NULL.
 #'
 #' @details Async recv will block while awaiting a message to arrive and is
 #'     usually the desired result. Set a timeout to allow the function to return
 #'     if no data is forthcoming.
 #'
-#'     In case of an error in unserialisation or data conversion, the function
-#'     will return the received raw vector to allow the data to be recovered.
+#'     In the event of an error in unserialisation or data conversion, or in the
+#'     evaluation of the function with respect to the data, a NULL byte (or
+#'     serialized NULL byte) will be sent in reply to signal an error to the client.
 #'
 #' @examples
 #' req <- socket("req", listen = "tcp://127.0.0.1:6546")
@@ -211,11 +212,11 @@ ctx_recv <- function(context,
 #'
 ctx_rep <- function(context,
                     execute,
-                    ...,
                     recv_mode = c("serial", "character", "complex", "double",
                                   "integer", "logical", "numeric", "raw"),
                     send_mode = c("serial", "raw"),
-                    timeout) {
+                    timeout,
+                    ...) {
 
   recv_mode <- match.arg(recv_mode)
   send_mode <- match.arg(send_mode)
@@ -225,14 +226,14 @@ ctx_rep <- function(context,
     message(res, " : ", nng_error(res))
     return(invisible(res))
   }
-  on.exit(expr = return(res))
+  on.exit(expr = send_aio(context, writeBin(object = "", con = raw()), mode = send_mode))
   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(expr = NULL)
   msg <- execute(data, ...)
+  on.exit(expr = NULL)
   ctx_send(context, data = msg, mode = send_mode, timeout = timeout, echo = FALSE)
 
 }
@@ -260,6 +261,9 @@ ctx_rep <- function(context,
 #'     In case of an error in unserialisation or data conversion, the function
 #'     will return the received raw vector to allow the data to be recovered.
 #'
+#'     If an error occured in the server process, a NULL byte will be received
+#'     (as \code{$data} if 'recv_mode' = 'serial', as \code{$raw} otherwise).
+#'
 #' @examples
 #' req <- socket("req", listen = "tcp://127.0.0.1:6546")
 #' rep <- socket("rep", dial = "tcp://127.0.0.1:6546")

R/docs.R

@@ -122,7 +122,8 @@ NULL
 #'     tries hard to avoid copying data, and thus is very light-weight.
 #'
 #'     [\strong{URI, inproc://}] This transport uses URIs using the scheme inproc://,
-#'     followed by an arbitrary string of text, terminated by a NUL byte.
+#'     followed by an arbitrary string of text, terminated by a NULL byte.
+#'     inproc://nanonext is a valid example URL.
 #'
 #'     \itemize{
 #'     \item Multiple URIs can be used within the same application, and they will
@@ -147,10 +148,14 @@ NULL
 #'     followed by a path name in the file system where the socket or named pipe
 #'     should be created.
 #'     \itemize{
+#'     \item On POSIX platforms, the path is taken literally, and is relative to
+#'     the current directory, unless it begins with /, in which case it is
+#'     relative to the root directory. For example, ipc://nanonext refers to the
+#'     name nanonext in the current directory, whereas ipc:///tmp/nanonext
+#'     refers to nanonext located in /tmp.
 #'     \item On Windows, all names are prefixed by \\.\ pipe\ and do not reside
-#'     in the normal file system. On POSIX platforms, the path is taken literally,
-#'     and is relative to the current directory, unless it begins with /, in
-#'     which case it is relative to the root directory.
+#'     in the normal file system - the required prefix is added automatically
+#'     by NNG, so you should specify a URL such as ipc://nanonext directly.
 #'     }
 #'
 #'     \emph{UNIX Aliases}
@@ -166,8 +171,7 @@ NULL
 #'     [\strong{URI, abstract://}] On Linux, this transport also can support abstract
 #'     sockets. Abstract sockets use a URI-encoded name after the scheme,
 #'     which allows arbitrary values to be conveyed in the path, including
-#'     embedded NUL bytes. For example, the name "a\0b" would be represented as
-#'     abstract://a%00b.
+#'     embedded NULL bytes. abstract://nanonext is a valid example URL.
 #'
 #'     \itemize{
 #'     \item Abstract sockets do not have any representation in the file system,

R/utils.R

@@ -47,7 +47,7 @@ nng_error <- function(error) {
 
 #' Timer Utility
 #'
-#' Set a timer (stopwatch). Will print a message to the console (stdout) upon
+#' Set a timer (stopwatch). Will print a message to the console (stderr) upon
 #'     completion.
 #'
 #' @param time time in ms. Non-integer values will be translated to integer using