cleaner Aio structure and UI

Author: shikokuchuo

NAMESPACE

@@ -4,6 +4,8 @@ S3method("$",nano)
 S3method("[",nanoSocket)
 S3method("[[",nano)
 S3method(.DollarNames,nano)
+S3method(.DollarNames,recvAio)
+S3method(.DollarNames,sendAio)
 S3method(close,nanoContext)
 S3method(close,nanoDialer)
 S3method(close,nanoListener)

R/aio.R

@@ -46,9 +46,9 @@
 #' call_aio(res)
 #' res$result
 #'
-#' res <- recv_aio(s2, timeout = 100)
-#' res
-#' call_aio(res)$data
+#' msg <- recv_aio(s2, timeout = 100)
+#' msg
+#' call_aio(msg)$data
 #'
 #' close(s1)
 #' close(s2)
@@ -62,8 +62,8 @@ call_aio <- function(aio) {
   if (inherits(aio, "recvAio")) {
 
     res <- .Call(rnng_aio_wait_get_msg, .subset2(aio, "aio"))
-    mode <- .subset2(aio, "callparams")[[1L]]
-    keep.raw <- .subset2(aio, "callparams")[[2L]]
+    mode <- .subset2(aio, "mode")
+    keep.raw <- .subset2(aio, "keep.raw")
     is.integer(res) && {
       if (keep.raw) {
         rm("raw", envir = aio)
@@ -79,7 +79,6 @@ call_aio <- function(aio) {
       rm("data", envir = aio)
       `[[<-`(aio, "data", res)
       rm("aio", envir = aio)
-      rm("callparams", envir = aio)
       return(invisible(aio))
     })
     data <- switch(mode,
@@ -95,7 +94,6 @@ call_aio <- function(aio) {
     rm("data", envir = aio)
     `[[<-`(aio, "data", data)
     rm("aio", envir = aio)
-    rm("callparams", envir = aio)
 
   } else if (inherits(aio, "sendAio")) {
 

R/context.R

@@ -324,7 +324,8 @@ request <- function(context,
     return(invisible(aio))
   }
   env <- `class<-`(new.env(), "recvAio")
-  `[[<-`(env, "callparams", list(recv_mode, keep.raw))
+  `[[<-`(env, "mode", recv_mode)
+  `[[<-`(env, "keep.raw", keep.raw)
   data <- raw <- resolv <- NULL
   if (keep.raw) {
     makeActiveBinding(sym = "raw", fun = function(x) {

R/nano.R

@@ -216,7 +216,8 @@ print.nanoListener <- function(x, ...) {
 #'
 print.recvAio <- function(x, ...) {
 
-  cat("< recvAio >\n - $raw for raw message\n - $data for message data\n", file = stdout())
+  cat("< recvAio >\n - $data for message data\n",
+      if (.subset2(x, "keep.raw")) "- $raw for raw message\n", file = stdout())
   invisible(x)
 
 }
@@ -262,6 +263,22 @@ print.sendAio <- function(x, ...) {
 
 }
 
+#' @export
+#'
+.DollarNames.recvAio <- function(x, pattern = "") {
+
+  grep(pattern, c("data", if (.subset2(x, "keep.raw")) "raw"), value = TRUE, fixed = TRUE)
+
+}
+
+#' @export
+#'
+.DollarNames.sendAio <- function(x, pattern = "") {
+
+  grep(pattern, "result", value = TRUE, fixed = TRUE)
+
+}
+
 #' @export
 #'
 print.unresolvedValue <- function(x, ...) {

R/sendrecv.R

@@ -200,13 +200,13 @@ recv <- function(socket,
 #' send_aio(s1, data.frame(a = 1, b = 2), timeout = 100)
 #' res <- recv_aio(s2, timeout = 100, keep.raw = FALSE)
 #' res
-#' call_aio(res)
-#' res
+#' res$data
 #'
 #' send_aio(s1, c(1.1, 2.2, 3.3), mode = "raw", timeout = 100)
 #' res <- recv_aio(s2, mode = "double", timeout = 100)
-#' call_aio(res)
 #' res
+#' res$raw
+#' res$data
 #'
 #' send_aio(s1, "example message", mode = "raw", timeout = 100)
 #' res <- recv_aio(s2, mode = "character", timeout = 100)
@@ -234,7 +234,8 @@ recv_aio <- function(socket,
     return(invisible(aio))
   }
   env <- `class<-`(new.env(), "recvAio")
-  `[[<-`(env, "callparams", list(mode, keep.raw))
+  `[[<-`(env, "mode", mode)
+  `[[<-`(env, "keep.raw", keep.raw)
   data <- raw <- resolv <- NULL
   if (keep.raw) {
     makeActiveBinding(sym = "raw", fun = function(x) {

README.Rmd

@@ -49,9 +49,10 @@ Implemented transports:
 4. [Async and Concurrency](#async-and-concurrency)
 5. [RPC and Distributed Computing](#rpc-and-distributed-computing)
 6. [Publisher / Subscriber Model](#publisher-subscriber-model)
-7. [ncurl Minimalist http Client](#ncurl-minimalist-http-client)
-8. [Building from source](#building-from-source)
-9. [Links](#links)
+7. [Surveyor / Repondent Model](#surveyor-respondent-model)
+8. [ncurl Minimalist http Client](#ncurl-minimalist-http-client)
+9. [Building from source](#building-from-source)
+10. [Links](#links)
 
 ### Installation
 
@@ -275,26 +276,29 @@ In such a case, using `call_aio()` confirms that the operation has completed (or
 
 ### Publisher Subscriber Model
 
-{nanonext} fully implements NNG's pub/sub protocol as per the below example. 
+{nanonext} fully implements NNG's pub/sub protocol as per the below example.
 
 The built-in logging levels are also demonstrated here. NNG errors are always output to stderr and operation is otherwise silent by default. To enable key information events to be sent to stdout, use `logging(level = "info")`. 
 
 The log level can also be set externally in production environments via an environment variable `NANONEXT_LOG`.
 
 ```{r pub}
+
+# set logging level to include information events ------------------------------
 logging(level = "info")
 
 pub <- socket("pub", listen = "inproc://nanobroadcast")
 sub <- socket("sub", dial = "inproc://nanobroadcast")
 
+# subscribing to a specific topic 'examples' -----------------------------------
 sub |> subscribe(topic = "examples")
 pub |> send(c("examples", "this is an example"), mode = "raw", echo = FALSE)
 sub |> recv(mode = "character", keep.raw = FALSE)
 
 pub |> send(c("other", "this other topic will not be received"), mode = "raw", echo = FALSE)
 sub |> recv(mode = "character", keep.raw = FALSE)
 
-# specify NULL to subscribe to ALL topics
+# specify NULL to subscribe to ALL topics --------------------------------------
 sub |> subscribe(topic = NULL)
 pub |> send(c("newTopic", "this is a new topic"), mode = "raw", echo = FALSE)
 sub |> recv("character", keep.raw = FALSE)
@@ -303,17 +307,61 @@ sub |> unsubscribe(topic = NULL)
 pub |> send(c("newTopic", "this topic will now not be received"), mode = "raw", echo = FALSE)
 sub |> recv("character", keep.raw = FALSE)
 
-# however the topics explicitly subscribed to are still received
+# however the topics explicitly subscribed to are still received ---------------
 pub |> send(c("examples", "this example will still be received"), mode = "raw", echo = FALSE)
 sub |> recv(mode = "character", keep.raw = FALSE)
 
+# set logging level back to the default of errors only -------------------------
+logging(level = "error")
+
 close(pub)
 close(sub)
 
 ```
 
 [&laquo; Back to ToC](#table-of-contents)
 
+### Surveyor Respondent Model
+
+This type of topology is useful for applications such as service discovery.
+
+```{r survey}
+
+sur <- socket("surveyor", listen = "inproc://nanoservice")
+res1 <- socket("respondent", dial = "inproc://nanoservice")
+res2 <- socket("respondent", dial = "inproc://nanoservice")
+
+# sur sets a survey timeout, applying to this and subsequent surveys -----------
+sur |> survey_time(500)
+
+# sur sends a message and then requests 2 async receives -----------------------
+sur |> send("service check", echo = FALSE)
+aio1 <- sur |> recv_aio()
+aio2 <- sur |> recv_aio()
+
+# res1 receives the message and replies using an aio send function -------------
+res1 |> recv(keep.raw = FALSE)
+res1 |> send_aio("res1")
+
+# res2 receives the message but fails to reply ---------------------------------
+res2 |> recv(keep.raw = FALSE)
+
+# checking the aio - only the first will have resolved -------------------------
+aio1$data
+aio2$data
+
+# after the survey expires, the second resolves into a timeout error -----------
+Sys.sleep(0.5)
+aio2$data
+
+close(sur)
+close(res1)
+close(res2)
+
+```
+
+[&laquo; Back to ToC](#table-of-contents)
+
 ### ncurl Minimalist http Client
 
 `ncurl()` is a minimalistic http(s) client. In normal use, it takes only one argument, the URL. It can follow redirects.

README.md

@@ -49,9 +49,10 @@ Implemented transports:
 4.  [Async and Concurrency](#async-and-concurrency)
 5.  [RPC and Distributed Computing](#rpc-and-distributed-computing)
 6.  [Publisher / Subscriber Model](#publisher-subscriber-model)
-7.  [ncurl Minimalist http Client](#ncurl-minimalist-http-client)
-8.  [Building from source](#building-from-source)
-9.  [Links](#links)
+7.  [Surveyor / Repondent Model](#surveyor-respondent-model)
+8.  [ncurl Minimalist http Client](#ncurl-minimalist-http-client)
+9.  [Building from source](#building-from-source)
+10. [Links](#links)
 
 ### Installation
 
@@ -253,8 +254,8 @@ msg <- recv_aio(s2)
 call_aio(msg)
 msg
 #> < recvAio >
-#>  - $raw for raw message
 #>  - $data for message data
+#>  - $raw for raw message
 msg$data
 #>   a b
 #> 1 1 2
@@ -325,10 +326,9 @@ call_aio(aio)
 
 aio
 #> < recvAio >
-#>  - $raw for raw message
 #>  - $data for message data
 str(aio$data)
-#>  num [1:100000000] 0.837 1.301 2.225 1.43 -1.543 ...
+#>  num [1:100000000] -2.12 -0.152 -0.485 1.971 0.061 ...
 ```
 
 In this example the calculation is returned, but other operations may
@@ -354,48 +354,100 @@ The log level can also be set externally in production environments via
 an environment variable `NANONEXT_LOG`.
 
 ``` r
+# set logging level to include information events ------------------------------
 logging(level = "info")
-#> 2022-03-03 10:51:39 [ log level ] set to: info
+#> 2022-03-03 16:31:20 [ log level ] set to: info
 
 pub <- socket("pub", listen = "inproc://nanobroadcast")
-#> 2022-03-03 10:51:39 [ sock open ] id: 9 | protocol: pub 
-#> 2022-03-03 10:51:39 [ list start ] sock: 9 | url: 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
 sub <- socket("sub", dial = "inproc://nanobroadcast")
-#> 2022-03-03 10:51:39 [ sock open ] id: 10 | protocol: sub 
-#> 2022-03-03 10:51:39 [ dial start ] sock: 10 | url: 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
 
+# subscribing to a specific topic 'examples' -----------------------------------
 sub |> subscribe(topic = "examples")
-#> 2022-03-03 10:51:39 [ subscribe ] sock: 10 | topic: examples
+#> 2022-03-03 16:31:20 [ 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 10:51:39 [ 8 ] Try again
+#> 2022-03-03 16:31:20 [ 8 ] Try again
 
-# specify NULL to subscribe to ALL topics
+# specify NULL to subscribe to ALL topics --------------------------------------
 sub |> subscribe(topic = NULL)
-#> 2022-03-03 10:51:39 [ subscribe ] sock: 10 | topic: ALL
+#> 2022-03-03 16:31:20 [ 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 10:51:39 [ unsubscribe ] sock: 10 | topic: ALL
+#> 2022-03-03 16:31:20 [ 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 10:51:39 [ 8 ] Try again
+#> 2022-03-03 16:31:20 [ 8 ] Try again
 
-# however the topics explicitly subscribed to are still received
+# however the topics explicitly subscribed to are still received ---------------
 pub |> send(c("examples", "this example will still be received"), mode = "raw", echo = FALSE)
 sub |> recv(mode = "character", keep.raw = FALSE)
 #> [1] "examples"                            "this example will still be received"
 
+# set logging level back to the default of errors only -------------------------
+logging(level = "error")
+#> 2022-03-03 16:31:20 [ log level ] set to: error
+
 close(pub)
-#> 2022-03-03 10:51:39 [ sock close ] id: 9 | protocol: pub
 close(sub)
-#> 2022-03-03 10:51:39 [ sock close ] id: 10 | protocol: sub
+```
+
+[« Back to ToC](#table-of-contents)
+
+### Surveyor Respondent Model
+
+This type of topology is useful for applications such as service
+discovery.
+
+``` r
+sur <- socket("surveyor", listen = "inproc://nanoservice")
+res1 <- socket("respondent", dial = "inproc://nanoservice")
+res2 <- socket("respondent", dial = "inproc://nanoservice")
+
+# sur sets a survey timeout, applying to this and subsequent surveys -----------
+sur |> survey_time(500)
+
+# sur sends a message and then requests 2 async receives -----------------------
+sur |> send("service check", echo = FALSE)
+aio1 <- sur |> recv_aio()
+aio2 <- sur |> recv_aio()
+
+# res1 receives the message and replies using an aio send function -------------
+res1 |> recv(keep.raw = FALSE)
+#> [1] "service check"
+res1 |> send_aio("res1")
+#> < sendAio >
+#>  - $result for send result
+
+# res2 receives the message but fails to reply ---------------------------------
+res2 |> recv(keep.raw = FALSE)
+#> [1] "service check"
+
+# checking the aio - only the first will have resolved -------------------------
+aio1$data
+#> [1] "res1"
+aio2$data
+#> < unresolved value >
+
+# 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
+
+close(sur)
+close(res1)
+close(res2)
 ```
 
 [« Back to ToC](#table-of-contents)
@@ -411,11 +463,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 39 64 62 62 2d 34 61 61 66 66 65 31 37 32 61 64 64 35 39 35
-#> [101] 35 37 33 31 62 35 34 30 30 22 0a 20 20 7d 0a 7d 0a
+#>  [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
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62209dbb-4aaffe172add5955731b5400\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-6220ed59-21d6910212b0c94d0ea62c10\"\n  }\n}\n"
 ```
 
 For advanced use, supports additional HTTP methods such as POST or PUT.