tidy Aio functions, introduce 'errorValue'

Author: shikokuchuo

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: nanonext
 Type: Package
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 0.2.0.9004
+Version: 0.2.0.9005
 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

@@ -10,6 +10,7 @@ S3method(close,nanoContext)
 S3method(close,nanoDialer)
 S3method(close,nanoListener)
 S3method(close,nanoSocket)
+S3method(print,errorValue)
 S3method(print,nanoContext)
 S3method(print,nanoDialer)
 S3method(print,nanoListener)

NEWS.md

@@ -1,8 +1,9 @@
-# nanonext 0.2.0.9004 (development)
+# nanonext 0.2.0.9005 (development)
 
 #### New Features
 
 * Aio values `$result`, `$raw` or `$data` now resolve without requiring `call_aio()`. Access the values directly and an NA 'unresolved value' will be returned if the Aio operation is yet to complete.
+* Integer error values generated by all receive functions are now classed 'errorValue' to be immediately distinguishable from possible message values.
 * `unresolved()` added as an auxiliary function to query whether an Aio is unresolved, for use in control flow statements.
 * `is_nul_byte()` added as a helper function for request/reply setups.
 * `survey_time()` added as a convenience function for surveyor/respondent patterns.

R/aio.R

@@ -57,54 +57,12 @@
 #'
 call_aio <- function(aio) {
 
-  is.null(.subset2(aio, "aio")) && return(invisible(aio))
-
+  .Call(rnng_aio_call, .subset2(aio, "aio")) && return(invisible(aio))
   if (inherits(aio, "recvAio")) {
-
-    res <- .Call(rnng_aio_wait_get_msg, .subset2(aio, "aio"))
-    mode <- .subset2(aio, "mode")
-    keep.raw <- .subset2(aio, "keep.raw")
-    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 = {
-      if (keep.raw) rm("raw", envir = aio)
-      rm("data", envir = aio)
-      `[[<-`(aio, "data", res)
-      rm("aio", envir = aio)
-      return(invisible(aio))
-    })
-    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()
-    if (keep.raw) {
-      rm("raw", envir = aio)
-      `[[<-`(aio, "raw", res)
-    }
-    rm("data", envir = aio)
-    `[[<-`(aio, "data", data)
-    rm("aio", envir = aio)
-
+    .subset2(aio, "data")
   } else if (inherits(aio, "sendAio")) {
-
-    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))
-
+    .subset2(aio, "result")
   }
-
   invisible(aio)
 
 }

R/context.R

@@ -145,7 +145,7 @@ recv_ctx <- function(context,
   res <- .Call(rnng_ctx_recv, context, timeout)
   is.integer(res) && {
     message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-    return(invisible(res))
+    return(invisible(`class<-`(res, "errorValue")))
   }
   on.exit(expr = return(res))
   data <- switch(mode,
@@ -231,7 +231,7 @@ reply <- function(context,
   res <- .Call(rnng_ctx_recv, context, timeout)
   is.integer(res) && {
     message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-    return(invisible(res))
+    return(invisible(`class<-`(res, "errorValue")))
   }
   on.exit(expr = send_aio(context, as.raw(0L), mode = send_mode))
   data <- switch(recv_mode,
@@ -321,21 +321,19 @@ request <- function(context,
   aio <- .Call(rnng_recv_aio, context, timeout)
   is.integer(aio) && {
     message(Sys.time(), " [ ", aio, " ] ", nng_error(aio))
-    return(invisible(aio))
+    return(invisible(`class<-`(aio, "errorValue")))
   }
   env <- `class<-`(new.env(), "recvAio")
-  `[[<-`(env, "mode", recv_mode)
-  `[[<-`(env, "keep.raw", 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
+          data <<- raw <<- resolv <<- `class<-`(res, "errorValue")
           message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-          return(invisible(res))
+          return(invisible(resolv))
         }
         on.exit(expr = {
           raw <<- res
@@ -360,9 +358,9 @@ request <- function(context,
       res <- .Call(rnng_aio_get_msg, aio)
       missing(res) && return(.Call(rnng_aio_unresolv))
       is.integer(res) && {
-        res <<- raw <<- resolv <<- res
+        data <<- raw <<- resolv <<- `class<-`(res, "errorValue")
         message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-        return(invisible(res))
+        return(invisible(resolv))
       }
       on.exit(expr = {
         data <<- res
@@ -381,6 +379,7 @@ request <- function(context,
     }
     data
   }, env = env)
+  `[[<-`(env, "keep.raw", keep.raw)
   `[[<-`(env, "aio", aio)
 
 }

R/nano.R

@@ -288,3 +288,12 @@ print.unresolvedValue <- function(x, ...) {
 
 }
 
+#' @export
+#'
+print.errorValue <- function(x, ...) {
+
+  cat("'errorValue' int", x, "\n", file = stdout())
+  invisible(x)
+
+}
+

R/sendrecv.R

@@ -95,13 +95,13 @@ send_aio <- function(socket, data, mode = c("serial", "raw"), timeout) {
     return(invisible(aio))
   }
   env <- `class<-`(new.env(), "sendAio")
-  result <- resolv <- NULL
+  result <- NULL
   makeActiveBinding(sym = "result", fun = function(x) {
-    if (is.null(resolv)) {
-      result <- .Call(rnng_aio_result, aio)
-      missing(result) && return(.Call(rnng_aio_unresolv))
-      result <<- result
-      resolv <<- 0L
+    if (is.null(result)) {
+      res <- .Call(rnng_aio_result, aio)
+      missing(res) && return(.Call(rnng_aio_unresolv))
+      if (res) message(Sys.time(), " [ ", res, " ] ", nng_error(res))
+      result <<- res
     }
     result
   }, env = env)
@@ -163,7 +163,7 @@ recv <- function(socket,
   res <- .Call(rnng_recv, socket, block)
   is.integer(res) && {
     message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-    return(invisible(res))
+    return(invisible(`class<-`(res, "errorValue")))
   }
   on.exit(expr = return(res))
   data <- switch(mode,
@@ -231,21 +231,19 @@ recv_aio <- function(socket,
   aio <- .Call(rnng_recv_aio, socket, timeout)
   is.integer(aio) && {
     message(Sys.time(), " [ ", aio, " ] ", nng_error(aio))
-    return(invisible(aio))
+    return(invisible(`class<-`(aio, "errorValue")))
   }
   env <- `class<-`(new.env(), "recvAio")
-  `[[<-`(env, "mode", mode)
-  `[[<-`(env, "keep.raw", 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
+          data <<- raw <<- resolv <<- `class<-`(res, "errorValue")
           message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-          return(invisible(res))
+          return(invisible(resolv))
         }
         on.exit(expr = {
           raw <<- res
@@ -270,9 +268,9 @@ recv_aio <- function(socket,
       res <- .Call(rnng_aio_get_msg, aio)
       missing(res) && return(.Call(rnng_aio_unresolv))
       is.integer(res) && {
-        res <<- raw <<- resolv <<- res
+        data <<- raw <<- resolv <<- `class<-`(res, "errorValue")
         message(Sys.time(), " [ ", res, " ] ", nng_error(res))
-        return(invisible(res))
+        return(invisible(resolv))
       }
       on.exit(expr = {
         data <<- res
@@ -291,6 +289,7 @@ recv_aio <- function(socket,
     }
     data
   }, env = env)
+  `[[<-`(env, "keep.raw", keep.raw)
   `[[<-`(env, "aio", aio)
 
 }

README.Rmd

@@ -323,7 +323,9 @@ close(sub)
 
 ### Surveyor Respondent Model
 
-This type of topology is useful for applications such as service discovery.
+This type of pattern is useful for applications such as service discovery.
+
+A surveyor sends a survey, which is broadcast to all peer respondents. The respondents then have a chance to reply (but are not obliged to). The survey itself is a timed event, such that responses received after the timeout are discarded.
 
 ```{r survey}
 

README.md

@@ -328,7 +328,7 @@ aio
 #> < recvAio >
 #>  - $data for message data
 str(aio$data)
-#>  num [1:100000000] -2.12 -0.152 -0.485 1.971 0.061 ...
+#>  num [1:100000000] -1.211 -0.218 0.105 -0.608 0.256 ...
 ```
 
 In this example the calculation is returned, but other operations may
@@ -356,38 +356,38 @@ an environment variable `NANONEXT_LOG`.
 ``` r
 # set logging level to include information events ------------------------------
 logging(level = "info")
-#> 2022-03-03 16:31:20 [ log level ] set to: info
+#> 2022-03-03 23:42:57 [ log level ] set to: info
 
 pub <- socket("pub", listen = "inproc://nanobroadcast")
-#> 2022-03-03 16:31:20 [ sock open ] id: 9 | protocol: pub 
-#> 2022-03-03 16:31:20 [ list start ] sock: 9 | url: inproc://nanobroadcast
+#> 2022-03-03 23:42:57 [ sock open ] id: 9 | protocol: pub 
+#> 2022-03-03 23:42:57 [ list start ] sock: 9 | url: inproc://nanobroadcast
 sub <- socket("sub", dial = "inproc://nanobroadcast")
-#> 2022-03-03 16:31:20 [ sock open ] id: 10 | protocol: sub 
-#> 2022-03-03 16:31:20 [ dial start ] sock: 10 | url: inproc://nanobroadcast
+#> 2022-03-03 23:42:57 [ sock open ] id: 10 | protocol: sub 
+#> 2022-03-03 23:42:57 [ dial start ] sock: 10 | url: inproc://nanobroadcast
 
 # subscribing to a specific topic 'examples' -----------------------------------
 sub |> subscribe(topic = "examples")
-#> 2022-03-03 16:31:20 [ subscribe ] sock: 10 | topic: examples
+#> 2022-03-03 23:42:57 [ subscribe ] sock: 10 | topic: examples
 pub |> send(c("examples", "this is an example"), mode = "raw", echo = FALSE)
 sub |> recv(mode = "character", keep.raw = FALSE)
 #> [1] "examples"           "this is an example"
 
 pub |> send(c("other", "this other topic will not be received"), mode = "raw", echo = FALSE)
 sub |> recv(mode = "character", keep.raw = FALSE)
-#> 2022-03-03 16:31:20 [ 8 ] Try again
+#> 2022-03-03 23:42:57 [ 8 ] Try again
 
 # specify NULL to subscribe to ALL topics --------------------------------------
 sub |> subscribe(topic = NULL)
-#> 2022-03-03 16:31:20 [ subscribe ] sock: 10 | topic: ALL
+#> 2022-03-03 23:42:57 [ subscribe ] sock: 10 | topic: ALL
 pub |> send(c("newTopic", "this is a new topic"), mode = "raw", echo = FALSE)
 sub |> recv("character", keep.raw = FALSE)
 #> [1] "newTopic"            "this is a new topic"
 
 sub |> unsubscribe(topic = NULL)
-#> 2022-03-03 16:31:20 [ unsubscribe ] sock: 10 | topic: ALL
+#> 2022-03-03 23:42:57 [ unsubscribe ] sock: 10 | topic: ALL
 pub |> send(c("newTopic", "this topic will now not be received"), mode = "raw", echo = FALSE)
 sub |> recv("character", keep.raw = FALSE)
-#> 2022-03-03 16:31:20 [ 8 ] Try again
+#> 2022-03-03 23:42:57 [ 8 ] Try again
 
 # however the topics explicitly subscribed to are still received ---------------
 pub |> send(c("examples", "this example will still be received"), mode = "raw", echo = FALSE)
@@ -396,7 +396,7 @@ sub |> recv(mode = "character", keep.raw = FALSE)
 
 # set logging level back to the default of errors only -------------------------
 logging(level = "error")
-#> 2022-03-03 16:31:20 [ log level ] set to: error
+#> 2022-03-03 23:42:57 [ log level ] set to: error
 
 close(pub)
 close(sub)
@@ -406,9 +406,14 @@ close(sub)
 
 ### Surveyor Respondent Model
 
-This type of topology is useful for applications such as service
+This type of pattern is useful for applications such as service
 discovery.
 
+A surveyor sends a survey, which is broadcast to all peer respondents.
+The respondents then have a chance to reply (but are not obliged to).
+The survey itself is a timed event, such that responses received after
+the timeout are discarded.
+
 ``` r
 sur <- socket("surveyor", listen = "inproc://nanoservice")
 res1 <- socket("respondent", dial = "inproc://nanoservice")
@@ -442,8 +447,8 @@ aio2$data
 # after the survey expires, the second resolves into a timeout error -----------
 Sys.sleep(0.5)
 aio2$data
-#> 2022-03-03 16:31:21 [ 5 ] Timed out
-#> [1] 5
+#> 2022-03-03 23:42:57 [ 5 ] Timed out
+#> 'errorValue' int 5
 
 close(sur)
 close(res1)
@@ -463,11 +468,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 32 30 65 64 35 39 2d 32 31 64 36 39 31 30 32 31 32 62 30 63 39 34
-#> [101] 64 30 65 61 36 32 63 31 30 22 0a 20 20 7d 0a 7d 0a
+#>  [76] 2d 36 32 32 31 35 32 38 31 2d 37 65 32 32 39 65 31 62 36 65 30 31 62 63 38
+#> [101] 33 36 31 37 61 61 64 31 64 22 0a 20 20 7d 0a 7d 0a
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-6220ed59-21d6910212b0c94d0ea62c10\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62215281-7e229e1b6e01bc83617aad1d\"\n  }\n}\n"
 ```
 
 For advanced use, supports additional HTTP methods such as POST or PUT.