move encode down to C level

Author: shikokuchuo

R/aio.R

@@ -49,9 +49,8 @@ send_aio <- function(con, data, mode = c("serial", "raw"), timeout = -2L) UseMet
 #'
 send_aio.nanoSocket <- function(con, data, mode = c("serial", "raw"), timeout = -2L) {
 
-  mode <- match.arg2(mode, c("serial", "raw"))
-  force(data)
-  data <- encode(data = data, mode = mode)
+  if (missing(mode) || match.arg1(mode) == 1L)
+    data <- serialize(object = data, connection = NULL)
   aio <- .Call(rnng_send_aio, con, data, timeout)
   is.integer(aio) && return(aio)
 
@@ -78,9 +77,8 @@ send_aio.nanoSocket <- function(con, data, mode = c("serial", "raw"), timeout =
 #'
 send_aio.nanoContext <- function(con, data, mode = c("serial", "raw"), timeout = -2L) {
 
-  mode <- match.arg2(mode, c("serial", "raw"))
-  force(data)
-  data <- encode(data = data, mode = mode)
+  if (missing(mode) || match.arg1(mode) == 1L)
+    data <- serialize(object = data, connection = NULL)
   aio <- .Call(rnng_ctx_send_aio, con, data, timeout)
   is.integer(aio) && return(aio)
 
@@ -107,8 +105,6 @@ send_aio.nanoContext <- function(con, data, mode = c("serial", "raw"), timeout =
 #'
 send_aio.nanoStream <- function(con, data, mode = "raw", timeout = -2L) {
 
-  force(data)
-  data <- encode(data = data, mode = 2L)
   aio <- .Call(rnng_stream_send_aio, con, data, timeout)
   is.integer(aio) && return(aio)
 
@@ -479,7 +475,7 @@ stop_aio <- function(aio) {
 #'
 #' while (unresolved(aio)) {
 #'   # do stuff before checking resolution again
-#'   cat("unresolved")
+#'   cat("unresolved\n")
 #'   s2 <- socket("pair", dial = "inproc://nanonext")
 #'   Sys.sleep(0.01)
 #' }

R/context.R

@@ -115,14 +115,16 @@ reply <- function(context,
 
   recv_mode <- match.arg2(recv_mode, c("serial", "character", "complex", "double",
                                        "integer", "logical", "numeric", "raw"))
-  send_mode <- match.arg2(send_mode, c("serial", "raw"))
   res <- .Call(rnng_ctx_recv, context, recv_mode, timeout, FALSE)
-  is_error_value(res) && return(res)
+  is_error_value(res) && return(invisible(res))
   on.exit(expr = send_aio(context, as.raw(0L), mode = send_mode))
   data <- execute(res, ...)
-  data <- encode(data = data, mode = send_mode)
+  if (missing(send_mode) || match.arg1(send_mode) == 1L)
+    data <- serialize(object = data, connection = NULL)
+  res <- .Call(rnng_ctx_send, context, data, timeout)
   on.exit()
-  invisible(.Call(rnng_ctx_send, context, data, timeout))
+  is.integer(res) && return(invisible(res))
+  invisible(0L)
 
 }
 
@@ -181,11 +183,10 @@ request <- function(context,
                     timeout = -2L,
                     keep.raw = TRUE) {
 
-  send_mode <- match.arg2(send_mode, c("serial", "raw"))
   recv_mode <- match.arg2(recv_mode, c("serial", "character", "complex", "double",
                                        "integer", "logical", "numeric", "raw"))
-  force(data)
-  data <- encode(data = data, mode = send_mode)
+  if (missing(send_mode) || match.arg1(send_mode) == 1L)
+    data <- serialize(object = data, connection = NULL)
   res <- .Call(rnng_ctx_send_aio, context, data, -2L)
   is.integer(res) && return(res)
 

R/messenger.R

@@ -55,8 +55,7 @@ messenger <- function(url) {
     data <- readline()
     if (identical(data, ":q")) break
     if (identical(data, "")) next
-    rdata <- writeBin(object = data, con = raw())
-    s <- suppressWarnings(.Call(rnng_send, sock, rdata, 0L))
+    s <- suppressWarnings(.Call(rnng_send, sock, data, 0L))
     if (is.integer(s)) {
       cat(sprintf("%*s > not sent: peer offline: %s\n", nchar(data), "", format.POSIXct(Sys.time())),
           file = stderr())

R/sendrecv.R

@@ -74,12 +74,11 @@ send.nanoSocket <- function(con,
                             block = FALSE,
                             echo = TRUE) {
 
-  mode <- match.arg2(mode, c("serial", "raw"))
-  force(data)
-  data <- encode(data = data, mode = mode)
+  if (missing(mode) || match.arg1(mode) == 1L)
+    data <- serialize(object = data, connection = NULL)
   res <- .Call(rnng_send, con, data, block)
-  res && return(invisible(res))
-  if (missing(echo) || isTRUE(echo)) data else invisible(res)
+  is.integer(res) && return(res)
+  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
 
 }
 
@@ -93,13 +92,12 @@ send.nanoContext <- function(con,
                              block = TRUE,
                              echo = TRUE) {
 
-  mode <- match.arg2(mode, c("serial", "raw"))
-  force(data)
-  data <- encode(data = data, mode = mode)
+  if (missing(mode) || match.arg1(mode) == 1L)
+    data <- serialize(object = data, connection = NULL)
   if (missing(block) || isTRUE(block)) block <- -2L
   res <- .Call(rnng_ctx_send, con, data, block)
-  res && return(invisible(res))
-  if (missing(echo) || isTRUE(echo)) data else invisible(res)
+  is.integer(res) && return(res)
+  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
 
 }
 
@@ -113,12 +111,10 @@ send.nanoStream <- function(con,
                             block = TRUE,
                             echo = TRUE) {
 
-  force(data)
-  data <- encode(data = data, mode = 2L)
   if (missing(block) || isTRUE(block)) block <- -2L
   res <- .Call(rnng_stream_send, con, data, block)
-  res && return(invisible(res))
-  if (missing(echo) || isTRUE(echo)) data else invisible(res)
+  is.integer(res) && return(res)
+  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
 
 }
 

R/utils.R

@@ -256,12 +256,6 @@ nano_init <- function(warn = c("immediate", "deferred", "error", "none")) {
 
 # nanonext - Non-exported functions --------------------------------------------
 
-encode <- function(data, mode) {
-  switch(mode,
-         serialize(object = data, connection = NULL),
-         if (is.raw(data)) data else writeBin(object = data, con = raw()))
-}
-
 match.arg2 <- function(choice, choices) {
   identical(choice, choices) && return(1L)
   index <- pmatch(choice[1L], choices, nomatch = 0L, duplicates.ok = TRUE)
@@ -270,3 +264,9 @@ match.arg2 <- function(choice, choices) {
   index
 }
 
+match.arg1 <- function(choice) {
+  index <- pmatch(choice[1L], c("serial", "raw"), nomatch = 0L, duplicates.ok = TRUE)
+  index || stop(sprintf("'%s' should be one of serial, raw", deparse(substitute(choice))))
+  index
+}
+

README.md

@@ -377,7 +377,7 @@ aio
 #> < recvAio >
 #>  - $data for message data
 aio$data |> str()
-#>  num [1:100000000] 0.369 -0.681 -0.413 0.83 -0.988 ...
+#>  num [1:100000000] -0.154 -0.281 -1.865 0.332 2.625 ...
 ```
 
 As `call_aio()` is blocking and will wait for completion, an alternative
@@ -514,11 +514,11 @@ ncurl("http://httpbin.org/headers")
 #>   [1] 7b 0a 20 20 22 68 65 61 64 65 72 73 22 3a 20 7b 0a 20 20 20 20 22 48 6f 73
 #>  [26] 74 22 3a 20 22 68 74 74 70 62 69 6e 2e 6f 72 67 22 2c 20 0a 20 20 20 20 22
 #>  [51] 58 2d 41 6d 7a 6e 2d 54 72 61 63 65 2d 49 64 22 3a 20 22 52 6f 6f 74 3d 31
-#>  [76] 2d 36 32 36 32 39 36 30 32 2d 34 38 64 39 33 64 39 33 33 64 36 32 35 66 30
-#> [101] 36 30 36 62 38 66 37 33 65 22 0a 20 20 7d 0a 7d 0a
+#>  [76] 2d 36 32 36 33 32 39 65 32 2d 34 61 64 37 64 36 33 62 35 64 33 62 30 65 63
+#> [101] 62 31 38 63 66 38 32 31 38 22 0a 20 20 7d 0a 7d 0a
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62629602-48d93d933d625f0606b8f73e\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-626329e2-4ad7d63b5d3b0ecb18cf8218\"\n  }\n}\n"
 ```
 
 For advanced use, supports additional HTTP methods such as POST or PUT.
@@ -533,7 +533,7 @@ res
 #>  - $raw for raw message
 
 call_aio(res)$data
-#> [1] "{\n  \"args\": {}, \n  \"data\": \"{\\\"key\\\": \\\"value\\\"}\", \n  \"files\": {}, \n  \"form\": {}, \n  \"headers\": {\n    \"Authorization\": \"Bearer APIKEY\", \n    \"Content-Length\": \"16\", \n    \"Content-Type\": \"application/json\", \n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62629602-1fb9951a0f4246df56d74460\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"78.145.225.121\", \n  \"url\": \"http://httpbin.org/post\"\n}\n"
+#> [1] "{\n  \"args\": {}, \n  \"data\": \"{\\\"key\\\": \\\"value\\\"}\", \n  \"files\": {}, \n  \"form\": {}, \n  \"headers\": {\n    \"Authorization\": \"Bearer APIKEY\", \n    \"Content-Length\": \"16\", \n    \"Content-Type\": \"application/json\", \n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-626329e3-3ad612de0091f2db61bb42eb\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"78.145.225.121\", \n  \"url\": \"http://httpbin.org/post\"\n}\n"
 ```
 
 In this respect, it may be used as a performant and lightweight method

man/unresolved.Rd

@@ -446,12 +446,12 @@ SEXP rnng_send_aio(SEXP socket, SEXP data, SEXP timeout) {
 
   nng_socket *sock = (nng_socket *) R_ExternalPtrAddr(socket);
   const nng_duration dur = (nng_duration) Rf_asInteger(timeout);
-  unsigned char *dp = RAW(data);
-  const R_xlen_t xlen = Rf_xlength(data);
   nano_aio *saio = R_Calloc(1, nano_aio);
   nng_msg *msg;
   int xc;
-  SEXP aio;
+  SEXP aio, enc = nano_encode(data);
+  const R_xlen_t xlen = Rf_xlength(enc);
+  unsigned char *dp = RAW(enc);
 
   saio->type = SENDAIO;
 
@@ -499,12 +499,12 @@ SEXP rnng_ctx_send_aio(SEXP context, SEXP data, SEXP timeout) {
 
   nng_ctx *ctxp = (nng_ctx *) R_ExternalPtrAddr(context);
   const nng_duration dur = (nng_duration) Rf_asInteger(timeout);
-  unsigned char *dp = RAW(data);
-  const R_xlen_t xlen = Rf_xlength(data);
   nano_aio *saio = R_Calloc(1, nano_aio);
   nng_msg *msg;
   int xc;
-  SEXP aio;
+  SEXP aio, enc = nano_encode(data);
+  const R_xlen_t xlen = Rf_xlength(enc);
+  unsigned char *dp = RAW(enc);
 
   saio->type = SENDAIO;
 
@@ -553,13 +553,13 @@ SEXP rnng_stream_send_aio(SEXP stream, SEXP data, SEXP timeout) {
 
   nng_stream *sp = (nng_stream *) R_ExternalPtrAddr(stream);
   const nng_duration dur = (nng_duration) Rf_asInteger(timeout);
-  unsigned char *dp = RAW(data);
-  const R_xlen_t xlen = Rf_xlength(data);
   const int frames = *LOGICAL(Rf_getAttrib(stream, nano_TextframesSymbol));
   nano_aio *iaio = R_Calloc(1, nano_aio);
   nng_iov *iov = R_Calloc(1, nng_iov);
   int xc;
-  SEXP aio;
+  SEXP aio, enc = nano_encode(data);
+  const R_xlen_t xlen = Rf_xlength(enc);
+  unsigned char *dp = RAW(enc);
 
   iaio->type = IOV_SENDAIO;
   iaio->data = iov;