implement pub/sub for all vector types

Author: shikokuchuo

NEWS.md

@@ -7,8 +7,10 @@
 
 #### 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.
 * 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.
 * All-round internal performance optimisations.
 

R/protocol.R

@@ -6,19 +6,18 @@
 #'     topic to subscribe to.
 #'
 #' @param socket a Socket using the sub protocol.
-#' @param topic [default NULL] a topic (given as a character string). The default
-#'     NULL subscribes to all topics.
+#' @param topic [default NULL] a topic (vector or NULL). The default NULL
+#'     subscribes to all topics.
 #'
 #' @return Invisibly, an integer exit code (zero on success).
 #'
 #' @details To use pub/sub the publisher must:
 #'     \itemize{
 #'     \item{specify \code{mode = 'raw'} when sending to allow the topics to be
 #'     recognised by the receiving party.}
-#'     \item{send a vector that separates the topic from the rest of the message
-#'     e.g. \code{send(socket, c("topic", "message"), mode = "raw")} - this
-#'     ensures that topic ends with the required nul byte for it to be
-#'     recognised.}
+#'     \item{when sending a character vector, the topic must be separated from
+#'     the rest of the message using e.g.
+#'     \code{send(socket, c("topic", "message"), mode = "raw")}}
 #'     }
 #'
 #' @examples
@@ -38,7 +37,7 @@
 #'
 subscribe <- function(socket, topic = NULL) {
 
-  invisible(.Call(rnng_socket_set, socket, 5L, "sub:subscribe", topic))
+  invisible(.Call(rnng_socket_set, socket, 0L, "sub:subscribe", topic))
 
 }
 
@@ -48,8 +47,8 @@ subscribe <- function(socket, topic = NULL) {
 #'     a topic from the subscription list.
 #'
 #' @param socket a Socket using the sub protocol.
-#' @param topic [default NULL] a topic (given as a character string). The default
-#'     NULL unsubscribes from all topics (if all topics were previously subscribed).
+#' @param topic [default NULL] a topic (vector or NULL). The default NULL
+#'     unsubscribes from all topics (if all topics were previously subscribed).
 #'
 #' @return Invisibly, an integer exit code (zero on success).
 #'
@@ -60,10 +59,9 @@ subscribe <- function(socket, topic = NULL) {
 #'     \itemize{
 #'     \item{specify \code{mode = 'raw'} when sending to allow the topics to be
 #'     recognised by the receiving party.}
-#'     \item{send a vector that separates the topic from the rest of the message
-#'     e.g. \code{send(socket, c("topic", "message"), mode = "raw")} - this
-#'     ensures that topic ends with the required nul byte for it to be
-#'     recognised.}
+#'     \item{when sending a character vector, the topic must be separated from
+#'     the rest of the message using e.g.
+#'     \code{send(socket, c("topic", "message"), mode = "raw")}}
 #'     }
 #'
 #' @examples
@@ -84,7 +82,7 @@ subscribe <- function(socket, topic = NULL) {
 #'
 unsubscribe <- function(socket, topic = NULL) {
 
-  invisible(.Call(rnng_socket_set, socket, 5L, "sub:unsubscribe", topic))
+  invisible(.Call(rnng_socket_set, socket, 0L, "sub:unsubscribe", topic))
 
 }
 

README.md

@@ -105,7 +105,7 @@ Send message from ‘nano1’:
 
 ``` r
 nano1$send("hello world!")
-#>  [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
+#>  [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
 #> [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 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>  [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
 #> [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 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>  [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
 #> [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 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>  [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
 #> [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 02 00 00 03 05 00 00 00 00 05 55 54 46 2d 38 00 00
+#>   [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
 #>  [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.5354 0.1424 0.3993 0.0906 1.6356 ...
+#>  num [1:100000000] -0.901 0.2851 -0.0205 0.1886 1.3892 ...
 ```
 
 As `call_aio()` is blocking and will wait for completion, an alternative
@@ -528,11 +528,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 64 62 31 35 33 2d 32 39 39 31 38 39 63 35 33 61 30 30 35 61 32
-#> [101] 38 32 33 65 31 34 38 62 37 22 0a 20 20 7d 0a 7d 0a
+#>  [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
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-626db153-299189c53a005a2823e148b7\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-6273ae8e-731a6b5020ec601252bdf298\"\n  }\n}\n"
 ```
 
 For advanced use, supports additional HTTP methods such as POST or PUT.
@@ -548,7 +548,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-626db153-400a5a4e7880ba63075877ec\"\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-6273ae8f-5a31583a2f9f02b8619845b7\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"79.173.189.204\", \n  \"url\": \"http://httpbin.org/post\"\n}\n"
 ```
 
 In this respect, it may be used as a performant and lightweight method

man/subscribe.Rd

@@ -15,6 +15,14 @@ SEXP rnng_socket_set(SEXP socket, SEXP type, SEXP opt, SEXP value) {
   int xc;
 
   switch (typ) {
+  case 0:
+    if (value == R_NilValue) {
+      xc = nng_socket_set(*sock, op, NULL, 0);
+    } else {
+      SEXP enc = nano_encode(value);
+      xc = nng_socket_set(*sock, op, RAW(enc), Rf_xlength(enc));
+    }
+    break;
   case 1:
     xc = nng_socket_set_bool(*sock, op, (bool) Rf_asInteger(value));
     break;