add nanoContext methods for protocol helpers

Author: shikokuchuo

NAMESPACE

@@ -44,6 +44,12 @@ S3method(setopt,nanoSocket)
 S3method(setopt,nanoStream)
 S3method(start,nanoDialer)
 S3method(start,nanoListener)
+S3method(subscribe,nanoContext)
+S3method(subscribe,nanoSocket)
+S3method(survey_time,nanoContext)
+S3method(survey_time,nanoSocket)
+S3method(unsubscribe,nanoContext)
+S3method(unsubscribe,nanoSocket)
 export(.mirai_scm)
 export(call_aio)
 export(context)

NEWS.md

@@ -2,16 +2,17 @@
 
 #### New Features
 
-* `$context()` method added for creating new contexts from nano Objects using supported protocols (i.e. req, rep, surveyor, respondent) - this replaces the `context()` function for nano Objects.
+* `$context()` method added for creating new contexts from nano Objects using supported protocols (i.e. req, rep, sub, surveyor, respondent) - this replaces the `context()` function for nano Objects.
 * Added convenience auxiliary functions `is_nano()` and `is_aio()`.
+* `subscribe()` / `unsubscribe()` now accept a topic of any atomic type (not just character), allowing pub/sub to be used when sending integer, double, logical, complex, or raw vectors.
 
 #### Updates
 
-* `subscribe()` / `unsubscribe()` now accept a topic in any of the standard types (not just character), allowing pub/sub to be used when sending integer, double, logical, complex, or raw vectors.
+* Protocol-specific helpers `subscribe()`, `unsubscribe()`, and `survey_time()` gain nanoContext methods.
 * For receives, if an error occurs in unserialisation or data conversion (e.g. mode was incorrectly specified), the received raw vector is now available at both `$raw` and `$data` if `keep.raw = TRUE`.
 * Setting 'NANONEXT_TLS=1' now allows the downloaded NNG library to be built against a system mbedtls installation.
 * Setting 'NANONEXT_ARM' is no longer required on platforms such as Raspberry Pi - the package configure script should now detect platforms that require the libatomic linker flag to be set automatically.
-* Deprecated `send_ctx()`, `recv_ctx()` and logging are removed.
+* Deprecated `send_ctx()`, `recv_ctx()` and logging removed.
 * All-round internal performance optimisations.
 
 # nanonext 0.4.0

R/context.R

@@ -15,8 +15,8 @@
 #'     created on a rep socket can each receive requests, and send replies to
 #'     them, without any regard to or interference with each other.
 #'
-#'     Note: only certain protocols support creation of separate contexts -
-#'     namely req, rep, surveyor, respondent.
+#'     Only the following protocols support creation of contexts: req, rep, sub
+#'     (in a pub/sub pattern), surveyor, respondent.
 #'
 #'     To send and receive over a context use \code{\link{send}} and
 #'     \code{\link{recv}} or their async counterparts \code{\link{send_aio}} and

R/nano.R

@@ -134,6 +134,7 @@ nano <- function(protocol = c("pair", "bus", "req", "rep", "push", "pull",
            nano[["context"]] <- function() context(socket)
          },
          sub = {
+           nano[["context"]] <- function() context(socket)
            nano[["subscribe"]] <- function(topic = NULL) subscribe(socket,
                                                                    topic = topic)
            nano[["unsubscribe"]] <- function(topic = NULL) unsubscribe(socket,

R/protocol.R

@@ -2,11 +2,11 @@
 
 #' Subscribe Topic
 #'
-#' For a socket using the sub protocol in a publisher/subscriber pattern. Set a
-#'     topic to subscribe to.
+#' For a socket or context using the sub protocol in a publisher/subscriber
+#'     pattern. Set a topic to subscribe to.
 #'
-#' @param socket a Socket using the sub protocol.
-#' @param topic [default NULL] a topic (vector or NULL). The default NULL
+#' @param con a Socket or Context using the 'sub' protocol.
+#' @param topic [default NULL] an atomic type or NULL. The default NULL
 #'     subscribes to all topics.
 #'
 #' @return Invisibly, an integer exit code (zero on success).
@@ -30,24 +30,45 @@
 #' send(pub, c("other", "this other topic will not be received"), mode = "raw")
 #' recv(sub, "character")
 #'
+#' subscribe(sub, 2)
+#' send(pub, c(2, 10, 10, 20), mode = "raw")
+#' recv(sub, "double", keep.raw = FALSE)
+#'
 #' close(pub)
 #' close(sub)
 #'
+#' @rdname subscribe
+#' @export
+#'
+subscribe <- function(con, topic = NULL) UseMethod("subscribe")
+
+#' @rdname subscribe
+#' @method subscribe nanoSocket
 #' @export
 #'
-subscribe <- function(socket, topic = NULL) {
+subscribe.nanoSocket <- function(con, topic = NULL) {
 
-  invisible(.Call(rnng_socket_set, socket, 0L, "sub:subscribe", topic))
+  invisible(.Call(rnng_socket_set, con, 0L, "sub:subscribe", topic))
+
+}
+
+#' @rdname subscribe
+#' @method subscribe nanoContext
+#' @export
+#'
+subscribe.nanoContext <- function(con, topic = NULL) {
+
+  invisible(.Call(rnng_ctx_set, con, 0L, "sub:subscribe", topic))
 
 }
 
 #' Unsubscribe Topic
 #'
-#' For a socket using the sub protocol in a publisher/subscriber pattern. Remove
-#'     a topic from the subscription list.
+#' For a socket or context using the sub protocol in a publisher/subscriber
+#'     pattern. Remove a topic from the subscription list.
 #'
-#' @param socket a Socket using the sub protocol.
-#' @param topic [default NULL] a topic (vector or NULL). The default NULL
+#' @param con a Socket or Context using the 'sub' protocol.
+#' @param topic [default NULL] an atomic type or NULL. The default NULL
 #'     unsubscribes from all topics (if all topics were previously subscribed).
 #'
 #' @return Invisibly, an integer exit code (zero on success).
@@ -75,24 +96,46 @@ subscribe <- function(socket, topic = NULL) {
 #' send(pub, c("examples", "this example will not be received"), mode = "raw")
 #' recv(sub, "character")
 #'
+#' subscribe(sub, 2)
+#' send(pub, c(2, 10, 10, 20), mode = "raw")
+#' recv(sub, "double", keep.raw = FALSE)
+#'
 #' close(pub)
 #' close(sub)
 #'
+#' @rdname unsubscribe
 #' @export
 #'
-unsubscribe <- function(socket, topic = NULL) {
+unsubscribe <- function(con, topic = NULL) UseMethod("unsubscribe")
 
-  invisible(.Call(rnng_socket_set, socket, 0L, "sub:unsubscribe", topic))
+#' @rdname unsubscribe
+#' @method unsubscribe nanoSocket
+#' @export
+#'
+unsubscribe.nanoSocket <- function(con, topic = NULL) {
+
+  invisible(.Call(rnng_socket_set, con, 0L, "sub:unsubscribe", topic))
+
+}
+
+#' @rdname unsubscribe
+#' @method unsubscribe nanoContext
+#' @export
+#'
+unsubscribe.nanoContext <- function(con, topic = NULL) {
+
+  invisible(.Call(rnng_ctx_set, con, 0L, "sub:unsubscribe", topic))
 
 }
 
 #' Set Survey Time
 #'
-#' For a socket using the surveyor protocol in a surveyor/respondent pattern.
-#'     Set a survey timeout in ms (remains valid for all subsequent surveys).
-#'     Messages received by the surveyor after the timer has ended are discarded.
+#' For a socket or context using the surveyor protocol in a surveyor/respondent
+#'     pattern. Set a survey timeout in ms (remains valid for all subsequent
+#'     surveys). Messages received by the surveyor after the timer has ended are
+#'     discarded.
 #'
-#' @param socket a Socket or Context using the surveyor protocol.
+#' @param con a Socket or Context using the 'surveyor' protocol.
 #' @param time the survey timeout in ms.
 #'
 #' @return Invisibly, an integer exit code (zero on success).
@@ -126,11 +169,28 @@ unsubscribe <- function(socket, topic = NULL) {
 #' close(sur)
 #' close(res)
 #'
+#' @rdname survey_time
+#' @export
+#'
+survey_time <- function(con, time) UseMethod("survey_time")
+
+#' @rdname survey_time
+#' @method survey_time nanoSocket
+#' @export
+#'
+survey_time.nanoSocket <- function(con, time) {
+
+  invisible(.Call(rnng_socket_set, con, 3L, "surveyor:survey-time", time))
+
+}
+
+#' @rdname survey_time
+#' @method survey_time nanoContext
 #' @export
 #'
-survey_time <- function(socket, time) {
+survey_time.nanoContext <- function(con, time) {
 
-  invisible(.Call(rnng_socket_set, socket, 3L, "surveyor:survey-time", time))
+  invisible(.Call(rnng_ctx_set, con, 3L, "surveyor:survey-time", time))
 
 }
 

README.Rmd

@@ -348,6 +348,16 @@ sub |> recv("character", keep.raw = FALSE)
 pub |> send(c("examples", "this example will still be received"), mode = "raw", echo = FALSE)
 sub |> recv(mode = "character", keep.raw = FALSE)
 
+```
+
+The subscribed topic can be any atomic type (not just character), allowing integer, double, logical, complex and raw vectors to be sent as well.
+
+```{r pub2}
+
+sub |> subscribe(topic = 1)
+pub |> send(c(1, 10, 10, 20), mode = "raw", echo = FALSE)
+sub |> recv(mode = "double", keep.raw = FALSE)
+
 close(pub)
 close(sub)
 

README.md

@@ -105,7 +105,7 @@ Send message from ‘nano1’:
 
 ``` r
 nano1$send("hello world!")
-#>  [1] 58 0a 00 00 00 03 00 04 01 02 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>  [1] 58 0a 00 00 00 03 00 04 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
 #> [26] 00 10 00 00 00 01 00 04 00 09 00 00 00 0c 68 65 6c 6c 6f 20 77 6f 72 6c 64
 #> [51] 21
 ```
@@ -115,7 +115,7 @@ Receive message using ‘nano2’:
 ``` r
 nano2$recv()
 #> $raw
-#>  [1] 58 0a 00 00 00 03 00 04 01 02 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>  [1] 58 0a 00 00 00 03 00 04 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
 #> [26] 00 10 00 00 00 01 00 04 00 09 00 00 00 0c 68 65 6c 6c 6f 20 77 6f 72 6c 64
 #> [51] 21
 #> 
@@ -146,7 +146,7 @@ Send message from ‘socket1’:
 
 ``` r
 send(socket1, "hello world!")
-#>  [1] 58 0a 00 00 00 03 00 04 01 02 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>  [1] 58 0a 00 00 00 03 00 04 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
 #> [26] 00 10 00 00 00 01 00 04 00 09 00 00 00 0c 68 65 6c 6c 6f 20 77 6f 72 6c 64
 #> [51] 21
 ```
@@ -156,7 +156,7 @@ Receive message using ‘socket2’:
 ``` r
 recv(socket2)
 #> $raw
-#>  [1] 58 0a 00 00 00 03 00 04 01 02 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>  [1] 58 0a 00 00 00 03 00 04 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
 #> [26] 00 10 00 00 00 01 00 04 00 09 00 00 00 0c 68 65 6c 6c 6f 20 77 6f 72 6c 64
 #> [51] 21
 #> 
@@ -285,7 +285,7 @@ msg$data
 #>   a b
 #> 1 1 2
 msg$raw
-#>   [1] 58 0a 00 00 00 03 00 04 01 02 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>   [1] 58 0a 00 00 00 03 00 04 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
 #>  [26] 03 13 00 00 00 02 00 00 00 0e 00 00 00 01 3f f0 00 00 00 00 00 00 00 00 00
 #>  [51] 0e 00 00 00 01 40 00 00 00 00 00 00 00 00 00 04 02 00 00 00 01 00 04 00 09
 #>  [76] 00 00 00 05 6e 61 6d 65 73 00 00 00 10 00 00 00 02 00 04 00 09 00 00 00 01
@@ -388,7 +388,7 @@ aio
 #> < recvAio >
 #>  - $data for message data
 aio$data |> str()
-#>  num [1:100000000] -0.901 0.2851 -0.0205 0.1886 1.3892 ...
+#>  num [1:100000000] 0.173 -0.103 -0.522 -0.812 0.532 ...
 ```
 
 As `call_aio()` is blocking and will wait for completion, an alternative
@@ -446,6 +446,18 @@ sub |> recv("character", keep.raw = FALSE)
 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"
+```
+
+The subscribed topic can be any atomic type (not just character),
+allowing integer, double, logical, complex and raw vectors to be sent as
+well.
+
+``` r
+
+sub |> subscribe(topic = 1)
+pub |> send(c(1, 10, 10, 20), mode = "raw", echo = FALSE)
+sub |> recv(mode = "double", keep.raw = FALSE)
+#> [1]  1 10 10 20
 
 close(pub)
 close(sub)
@@ -528,11 +540,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 37 33 61 65 38 65 2d 37 33 31 61 36 62 35 30 32 30 65 63 36 30 31
-#> [101] 32 35 32 62 64 66 32 39 38 22 0a 20 20 7d 0a 7d 0a
+#>  [76] 2d 36 32 37 34 35 31 35 30 2d 32 61 62 61 34 61 31 61 36 39 63 37 36 61 66
+#> [101] 62 32 38 30 61 30 65 66 62 22 0a 20 20 7d 0a 7d 0a
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-6273ae8e-731a6b5020ec601252bdf298\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62745150-2aba4a1a69c76afb280a0efb\"\n  }\n}\n"
 ```
 
 For advanced use, supports additional HTTP methods such as POST or PUT.
@@ -548,7 +560,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-6273ae8f-5a31583a2f9f02b8619845b7\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"79.173.189.204\", \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-62745150-4a03a30d6fcd147d71c51fdf\"\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