Add context ID to request() serialized message header (#153)

* Encode ctx id in request() serialized message header

* Retain id argument compatibility

* Update test

Author: shikokuchuo

DESCRIPTION

@@ -38,6 +38,6 @@ Config/Needs/website: tidyverse/tidytemplate
 Config/usethis/last-upkeep: 2025-04-23
 Encoding: UTF-8
 Roxygen: list(markdown = TRUE)
-RoxygenNote: 7.3.2
+RoxygenNote: 7.3.2.9000
 SystemRequirements: 'libnng' >= 1.9 and 'libmbedtls' >= 2.5, or 'cmake'
     and 'xz' to compile NNG and/or Mbed TLS included in package sources

NEWS.md

@@ -2,6 +2,7 @@
 
 #### Updates
 
+* `request()` argument `id` is now a logical rather than integer argument.
 * The previous `listen()` and `dial()` argument `error`, removed in v1.6.0, is now defunct.
 * The previous `serial_config()` argument `vec`, unutilised since v1.6.0, is removed.
 

R/context.R

@@ -180,9 +180,9 @@ reply <- function(
 #' @param cv (optional) a 'conditionVariable' to signal when the async receive
 #'   is complete, or NULL. If any other value is supplied, this will cause the
 #'   pipe connection to be dropped when the async receive is complete.
-#' @param id (optional) integer message ID to send a special payload to the
-#'   context upon timeout (asynchronously) consisting of an integer zero,
-#'   followed by the value of `id` supplied.
+#' @param id (optional) specify as TRUE to send a message via the context upon
+#'   timeout (asynchronously) consisting of an integer zero, followed by the
+#'   integer context ID.
 #'
 #' @return A 'recvAio' (object of class 'mirai' and 'recvAio') (invisibly).
 #'

man/request.Rd

@@ -4,8 +4,8 @@
 
 // internals -------------------------------------------------------------------
 
+int special_header = 0;
 static int special_marker = 0;
-static int special_header = 0;
 static nano_serial_bundle nano_bundle;
 static SEXP nano_eval_res;
 

src/core.c

@@ -244,6 +244,7 @@ typedef struct nano_node_s {
   struct nano_node_s *next;
 } nano_node;
 
+extern int special_header;
 extern void (*eln2)(void (*)(void *), void *, double, int);
 
 extern SEXP nano_AioSymbol;

src/nanonext.h

@@ -408,19 +408,17 @@ SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeou
 
   if (NANO_PTR_CHECK(con, nano_ContextSymbol))
     Rf_error("`con` is not a valid Context");
+  nng_ctx *ctx = (nng_ctx *) NANO_PTR(con);
 
   const nng_duration dur = timeout == R_NilValue ? NNG_DURATION_DEFAULT : (nng_duration) nano_integer(timeout);
   const uint8_t mod = (uint8_t) nano_matcharg(recvmode);
   const int raw = nano_encode_mode(sendmode);
-  const int id = msgid != R_NilValue ? NANO_INTEGER(msgid) : 0;
-  int signal, drop, xc;
-  if (cvar == R_NilValue) {
-    signal = 0;
-    drop = 0;
-  } else {
-    signal = !NANO_PTR_CHECK(cvar, nano_CvSymbol);
-    drop = 1 - signal;
-  }
+  const int id = msgid == R_NilValue ? 0 : TYPEOF(msgid) == LGLSXP ? nng_ctx_id(*ctx) : NANO_INTEGER(msgid);
+  const int signal = cvar != R_NilValue && !NANO_PTR_CHECK(cvar, nano_CvSymbol);
+  const int drop = cvar != R_NilValue && !signal;
+  int xc;
+
+  nano_cv *ncv = signal ? (nano_cv *) NANO_PTR(cvar) : NULL;
 
   nano_saio *saio = NULL;
   nano_aio *raio = NULL;
@@ -431,6 +429,7 @@ SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeou
   if (raw) {
     nano_encode(&buf, data);
   } else {
+    special_header = id;
     nano_serialize(&buf, data, NANO_PROT(con));
   }
 
@@ -440,9 +439,6 @@ SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeou
   raio = calloc(1, sizeof(nano_aio));
   NANO_ENSURE_ALLOC(raio);
 
-  nng_ctx *ctx = (nng_ctx *) NANO_PTR(con);
-  nano_cv *ncv = signal ? (nano_cv *) NANO_PTR(cvar) : NULL;
-
   saio->ctx = ctx;
   saio->id = id;
 

src/sync.c

@@ -298,7 +298,7 @@ test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, cv = NA))
 test_notnull(cs$data)
 test_true(recv(ctxn, block = 500))
 test_zero(send(ctxn, TRUE, mode = 1L, block = 500))
-test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, timeout = 5, id = 12L))
+test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, timeout = 5, id = TRUE))
 test_zero(reap(ctxn))
 test_equal(reap(ctxn), 7L)
 test_zero(pipe_notify(rep, cv, add = TRUE, flag = TRUE))