implementing full AIO

Author: shikokuchuo

NAMESPACE

@@ -8,6 +8,7 @@ S3method(close,nanoContext)
 S3method(close,nanoDialer)
 S3method(close,nanoListener)
 S3method(close,nanoSocket)
+S3method(print,nano)
 S3method(print,nanoContext)
 S3method(print,nanoDialer)
 S3method(print,nanoListener)
@@ -19,13 +20,13 @@ S3method(setopt,nanoListener)
 S3method(setopt,nanoSocket)
 S3method(start,nanoDialer)
 S3method(start,nanoListener)
+export(aio_call)
+export(aio_stop)
 export(context)
 export(ctx_recv)
-export(ctx_recv_vec)
 export(ctx_rep)
 export(ctx_req)
 export(ctx_send)
-export(ctx_send_vec)
 export(dial)
 export(listen)
 export(nano)
@@ -35,11 +36,9 @@ export(nng_version)
 export(recv)
 export(recv_aio)
 export(recv_vec)
-export(recv_vec_aio)
 export(send)
 export(send_aio)
 export(send_vec)
-export(send_vec_aio)
 export(setopt)
 export(socket)
 export(subscribe)

NEWS.md

@@ -2,14 +2,15 @@
 
 #### New Features
 
-* New `ctx_rep()` and `ctx_req()` functions implement the full logic of an RPC server/client. Designed to be run in separate processes, the server will await data and apply an arbitrary function to it before returning a result, whilst the client will send data to the server and await a response.
+* Implemented full async I/O capabilities allowing `send_aio()` and `recv_aio()` to return Aio objects, for which the results may be called using `aio_call()`.
+* New `ctx_rep()` and `ctx_req()` functions implement the full logic of an RPC server/client. Designed to be run in separate processes, the server will await data and apply an arbitrary function before returning a result, whilst the client will send data to the server and await a response.
 * New `ncurl()` minimalistic http(s) client.
 * Allow setting the environment variable 'NANONEXT_TLS' prior to package installation to enable TLS where the system NNG library has been built with TLS support (using Mbed TLS).
 
 #### Updates
 
 * All send and receive functions, e.g. `send()`/`recv()`, gain a revamped 'mode' argument. This now permits the choice of whether to use R serialization, consolidating the functionality of the '_vec' series of functions.
-* The '_vec' series of functions ('send_vec', 'recv_vec' etc.) is deprecated and will be removed in a future release.
+* Functions 'send_vec' and 'recv_vec' are deprecated and will be removed in a future release.
 
 # nanonext 0.1.0
 

R/context.R

@@ -21,9 +21,9 @@
 #'
 #'     Note: not every protocol supports creation of separate contexts.
 #'
-#'     To send and receive over a context use \code{\link{ctx_send}} or
-#'     \code{\link{ctx_send_vec}} and \code{\link{ctx_recv}} or
-#'     \code{\link{ctx_recv_vec}} respectively.
+#'     To send and receive over a context use \code{\link{ctx_send}} and
+#'     \code{\link{ctx_recv}}. It is also possible to perform async send and receive
+#'     with a context using \code{\link{send_aio}} and \code{\link{recv_aio}}.
 #'
 #' @examples
 #' s <- socket("req", listen = "inproc://nanonext")
@@ -49,15 +49,15 @@ context <- function(socket) {
 
 }
 
-#' Send over Context (Async)
+#' Send over Context
 #'
-#' Send any number of R objects asynchronously over a Context, with the ability
-#'     to set send timeouts.
+#' Send data over a Context.
 #'
 #' @param context a Context.
+#' @inheritParams send
 #' @inheritParams send_aio
 #'
-#' @return A vector of zeros (invisibly) on success.
+#' @return Raw vector of sent data, or zero (invisibly) if 'echo' is set to FALSE.
 #'
 #' @details Will block if the send is in progress and has not yet completed -
 #'     certain protocol / transport combinations may limit the number of messages
@@ -69,41 +69,40 @@ context <- function(socket) {
 #' rep <- socket("rep", dial = "inproc://nanonext")
 #'
 #' ctx <- context(req)
-#' out <- ctx_send(ctx, data.frame(a = 1, b = 2), data.frame(c = 3, d = 4), timeout = 100)
-#' out
+#' ctx_send(ctx, data.frame(a = 1, b = 2), timeout = 100)
 #'
-#' msg <- recv_aio(rep, n = 2L, timeout = 100)
-#'
-#' out <- ctx_send(ctx, c(1.1, 2.2), c(3.3, 4.4), mode = "raw", timeout = 100)
-#' out
+#' msg <- recv_aio(rep, timeout = 100)
+#' ctx_send(ctx, c(1.1, 2.2, 3.3), mode = "raw", timeout = 100)
 #'
 #' close(req)
 #' close(rep)
 #'
 #' @export
 #'
-ctx_send <- function(context, ..., mode = c("serial", "raw"), timeout) {
+ctx_send <- function(context, data, mode = c("serial", "raw"), timeout, echo = TRUE) {
 
   mode <- match.arg(mode)
-  if (missing(timeout)) timeout <- -2L
+  force(data)
   data <- switch(mode,
-                 serial = lapply(list(...), serialize, connection = NULL),
-                 raw = lapply(list(...), writeBin, con = raw()))
+                 serial = serialize(object = data, connection = NULL),
+                 raw = if (is.raw(data)) data else writeBin(object = data, con = raw()))
   res <- .Call(rnng_ctx_send, context, data, timeout)
-  for (i in seq_along(res)) {
-    if (res[i]) message("[", i, "] ", res[i], " : ", nng_error(res[i]))
+  is.integer(res) && {
+    message(res, " : ", nng_error(res))
+    return(invisible(res))
   }
-  invisible(res)
+  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
 
 }
 
-#' Receive over Context (Async)
+#' Receive over Context
 #'
 #' Receive any number of R objects asynchronously over a Context, with the
 #'     ability to set receive timeouts.
 #'
 #' @param context a Context.
-#' @inheritParams recv_aio
+#' @inheritParams recv
+#' @inheritParams send_aio
 #'
 #' @return Named list of 2 elements: 'raw' containing a list of received raw
 #'     vectors and 'data' containing a list of converted R objects, or else a
@@ -126,37 +125,36 @@ ctx_send <- function(context, ..., mode = c("serial", "raw"), timeout) {
 #'
 #' ctxq <- context(req)
 #' ctxp <- context(rep)
-#' ctx_send(ctxq, data.frame(a = 1, b = 2), data.frame(c = 3, d = 4), timeout = 100)
-#' ctx_recv(ctxp, 2L, timeout = 100)
+#' ctx_send(ctxq, data.frame(a = 1, b = 2), timeout = 100)
+#' ctx_recv(ctxp, timeout = 100)
 #'
-#' ctx_send(ctxq, c(1.1, 2.2), c(3.3, 4.4), mode = "raw", timeout = 100)
-#' ctx_recv(ctxp, n = 2L, mode = "double", timeout = 100)
+#' ctx_send(ctxq, c(1.1, 2.2, 3.3), mode = "raw", timeout = 100)
+#' ctx_recv(ctxp, mode = "double", timeout = 100)
 #'
 #' close(req)
 #' close(rep)
 #'
 #' @export
 #'
 ctx_recv <- function(context,
-                     n = 1L,
                      mode = c("serial", "character", "complex", "double",
                               "integer", "logical", "numeric", "raw"),
                      timeout,
                      keep.raw = TRUE) {
 
   mode <- match.arg(mode)
   if (missing(timeout)) timeout <- -2L
-  res <- .Call(rnng_ctx_recv, context, n, timeout)
-  on.exit(expr = return(res))
-  data <- vector(mode = "list", length = length(res))
-  for (i in seq_along(res)) {
-    if (is.integer(res[[i]])) message("[", i, "] ", res[[i]], " : ", nng_error(res[[i]])) else
-      data[[i]] <- switch(mode,
-                          serial = unserialize(res[[i]]),
-                          character = (r <- readBin(con = res[[i]], what = mode, n = length(res[[i]])))[r != ""],
-                          raw = res[[i]],
-                          readBin(con = res[[i]], what = mode, n = length(res[[i]])))
+  res <- .Call(rnng_ctx_recv, context, timeout)
+  is.integer(res) && {
+    message(res, " : ", nng_error(res))
+    return(invisible(res))
   }
+  on.exit(expr = return(res))
+  data <- switch(mode,
+                 serial = unserialize(connection = res),
+                 character = (r <- readBin(con = res, what = mode, n = length(res)))[r != ""],
+                 raw = res,
+                 readBin(con = res, what = mode, n = length(res)))
   on.exit(expr = NULL)
   if (missing(keep.raw) || isTRUE(keep.raw)) list(raw = res, data = data) else data
 
@@ -203,11 +201,11 @@ ctx_recv <- function(context,
 #' ctxq <- context(req)
 #' ctxp <- context(rep)
 #'
-#' ctx_send(ctxq, 2022, timeout = 100)
+#' ctx_send(ctxq, 2022, timeout = 100, echo = FALSE)
 #' ctx_rep(ctxp, execute = function(x) x + 1, send_mode = "raw", timeout = 100)
 #' ctx_recv(ctxq, mode = "double", timeout = 100, keep.raw = FALSE)
 #'
-#' ctx_send(ctxq, 100, mode = "raw", timeout = 100)
+#' ctx_send(ctxq, 100, mode = "raw", timeout = 100, echo = FALSE)
 #' ctx_rep(ctxp, recv_mode = "double", execute = log, base = 10, timeout = 100)
 #' ctx_recv(ctxq, timeout = 100, keep.raw = FALSE)
 #'
@@ -227,7 +225,7 @@ ctx_rep <- function(context,
   recv_mode <- match.arg(recv_mode)
   send_mode <- match.arg(send_mode)
   if (missing(timeout)) timeout <- -2L
-  res <- .Call(rnng_ctx_recv, context, 1L, timeout)[[1L]]
+  res <- .Call(rnng_ctx_recv, context, timeout)
   is.integer(res) && {
     message(res, " : ", nng_error(res))
     return(invisible(res))
@@ -240,7 +238,7 @@ ctx_rep <- function(context,
                  readBin(con = res, what = recv_mode, n = length(res)))
   on.exit(expr = NULL)
   msg <- execute(data, ...)
-  ctx_send(context, msg, mode = send_mode, timeout = timeout)
+  ctx_send(context, data = msg, mode = send_mode, timeout = timeout, echo = FALSE)
 
 }
 
@@ -250,7 +248,7 @@ ctx_rep <- function(context,
 #'     data to the rep node (executor/server) and awaits the result to be returned.
 #'
 #' @inheritParams ctx_rep
-#' @inheritParams ctx_recv
+#' @inheritParams recv
 #' @param data an R object (if send_mode = 'raw', an R vector).
 #' @param timeout in ms. If unspecified, a socket-specific default timeout will
 #'     be used. Note this applies to each of the send and receive legs, hence the
@@ -294,15 +292,16 @@ ctx_req <- function(context,
   send_mode <- match.arg(send_mode)
   recv_mode <- match.arg(recv_mode)
   if (missing(timeout)) timeout <- -2L
-  data <- list(switch(send_mode,
-                      serial = serialize(object = data, connection = NULL),
-                      if (is.raw(data)) data else writeBin(object = data, con = raw())))
-  res <- .Call(rnng_ctx_send, context, data, timeout)[[1L]]
-  res == 0L || {
+  force(data)
+  data <- switch(send_mode,
+                 serial = serialize(object = data, connection = NULL),
+                 if (is.raw(data)) data else writeBin(object = data, con = raw()))
+  res <- .Call(rnng_ctx_send, context, data, timeout)
+  is.integer(res) && {
     message(res, " : ", nng_error(res))
     return(invisible(res))
   }
-  res <- .Call(rnng_ctx_recv, context, 1L, timeout)[[1L]]
+  res <- .Call(rnng_ctx_recv, context, timeout)
   is.integer(res) && {
     message(res, " : ", nng_error(res))
     return(invisible(res))
@@ -318,85 +317,3 @@ ctx_req <- function(context,
 
 }
 
-# Deprecated - may be removed at any time - do not use -------------------------
-
-#' Send Vector over Context (Async)
-#'
-#' DEPRECATED [Use ctx_send specifying mode = 'raw'] Send any number of R vectors
-#'     asynchronously over a Context, with the ability to set (optional) send
-#'     timeouts. Data will be sent as binary without R serialisation, hence
-#'     appropriate for interfacing with external programs.
-#'
-#' @inheritParams ctx_send
-#' @inheritParams send_vec_aio
-#'
-#' @return A vector of zeros (invisibly) on success.
-#'
-#' @details Will block if the send is in progress and has not yet completed -
-#'     certain protocol / transport combinations may limit the number of messages
-#'     that can be queued if they have yet to be received. Set a timeout to
-#'     ensure the function returns under all conditions.
-#'
-#' @keywords internal
-#' @export
-#'
-ctx_send_vec <- function(context, ..., timeout) {
-
-  if (missing(timeout)) timeout <- -2L
-  data <- lapply(list(...), writeBin, con = raw())
-  res <- .Call(rnng_ctx_send, context, data, timeout)
-  for (i in seq_along(res)) {
-    if (res[i]) message("[", i, "] ", res[i], " : ", nng_error(res[i]))
-  }
-  invisible(res)
-
-}
-
-#' Receive Vector over Context (Async)
-#'
-#' DEPRECATED [Use ctx_recv specifying mode] Receive vector data asynchronously
-#'     over a Context (with ability to set a timeout). The counterpart to
-#'     \code{\link{ctx_send_vec}}, data will be re-created from the raw vector
-#'     according to the specified mode. Can be used when interfacing with
-#'     external programs.
-#'
-#' @inheritParams ctx_recv
-#' @inheritParams recv_vec
-#'
-#' @return Named list of 2 elements: 'raw' containing a list of received raw
-#'     vectors and 'data' containing a list of vectors decoded to the type 'mode',
-#'     or else a list of vectors decoded to type 'mode' if keep.raw is set to
-#'     FALSE.
-#'
-#'     Note: a list of lists is always returned even when n = 1. To access the
-#'     first raw element, for example, use \code{$raw[[1]]} and the first data
-#'     element use \code{$data[[1]]}.
-#'
-#' @details Async recv will block while awaiting all 'n' messages to arrive. Set
-#'     a timeout to ensure that the function returns under all conditions.
-#'
-#' @keywords internal
-#' @export
-#'
-ctx_recv_vec <- function(context,
-                         mode = c("character", "complex", "double", "integer",
-                                  "logical", "numeric", "raw"),
-                         n = 1L,
-                         timeout,
-                         keep.raw = TRUE) {
-
-  mode <- match.arg(mode)
-  if (missing(timeout)) timeout <- -2L
-  res <- .Call(rnng_ctx_recv, context, n, timeout)
-  data <- vector(mode = "list", length = length(res))
-  for (i in seq_along(res)) {
-    if (is.integer(res[[i]])) message("[", i, "] ", res[[i]], " : ", nng_error(res[[i]])) else
-      data[[i]] <- switch(mode,
-                          character = (r <- readBin(con = res[[i]], what = mode, n = length(res[[i]])))[r != ""],
-                          raw = res[[i]],
-                          readBin(con = res[[i]], what = mode, n = length(res[[i]])))
-  }
-  if (missing(keep.raw) || isTRUE(keep.raw)) list(raw = res, data = data) else data
-
-}
-

R/nano.R

@@ -97,15 +97,8 @@ nano <- function(protocol = c("pair", "bus", "push", "pull", "req", "rep",
                                                    mode = mode,
                                                    block = block,
                                                    keep.raw = keep.raw)
-  nano[["recv_aio"]] <- function(n = 1L,
-                                 mode = c("serial", "character", "complex", "double",
-                                          "integer", "logical", "numeric", "raw"),
-                                 timeout,
-                                 keep.raw = TRUE) recv_aio(socket,
-                                                           n = n,
-                                                           mode = mode,
-                                                           timeout = timeout,
-                                                           keep.raw = keep.raw)
+  nano[["recv_aio"]] <- function(timeout) recv_aio(socket,
+                                                   timeout = timeout)
   nano[["send"]] <- function(data,
                              mode = c("serial", "raw"),
                              block = FALSE,
@@ -114,10 +107,10 @@ nano <- function(protocol = c("pair", "bus", "push", "pull", "req", "rep",
                                                mode = mode,
                                                block = block,
                                                echo = echo)
-  nano[["send_aio"]] <- function(...,
+  nano[["send_aio"]] <- function(data,
                                  mode = c("serial", "raw"),
                                  timeout) send_aio(socket,
-                                                   ...,
+                                                   data = data,
                                                    mode = mode,
                                                    timeout = timeout)
   nano[["socket_close"]] <- function() close(socket)
@@ -206,6 +199,18 @@ print.nanoListener <- function(x, ...) {
 
 }
 
+#' @export
+#'
+print.nano <- function(x, ...) {
+
+  klass <- attr(x, "class")[1L]
+  switch(klass,
+         recvAio = cat("< recvAio > - use aio_call() to retrieve message\n"),
+         sendAio = cat("< sendAio > - use aio_call() to retrieve result\n"))
+  invisible(x)
+
+}
+
 #' @export
 #'
 `[[.nano` <- function(x, i, exact = FALSE) {