request() msgid -> id

Author: shikokuchuo

NEWS.md

@@ -7,7 +7,7 @@
 * Adds `pipe_id()` for returning the integer pipe ID for a resolved 'recvAio'.
 * Adds `write_stdout()` which performs a non-buffered write to `stdout`, to avoid interleaved messages when used concurrently by different processes.
 * Adds `read_stdin()` which performs a read from `stdin` on a background thread, relayed via an 'inproc' socket so that it may be consumed via `recv()` or `recv_aio()`.
-* `request()` gains integer argument `msgid`. This may be specified to have a special payload sent asynchronously upon timeout (to communicate with the connected party).
+* `request()` gains integer argument `id`. This may be specified to have a special payload sent asynchronously upon timeout (to communicate with the connected party).
 
 #### Updates
 

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 msgid (optional) integer message ID to send a special payload to the
+#' @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 `msgid` supplied.
+#'   followed by the value of `id` supplied.
 #'
 #' @return A 'recvAio' (object of class 'mirai' and 'recvAio') (invisibly).
 #'
@@ -230,6 +230,6 @@ request <- function(
   recv_mode = c("serial", "character", "complex", "double", "integer", "logical", "numeric", "raw", "string"),
   timeout = NULL,
   cv = NULL,
-  msgid = NULL
+  id = NULL
 )
-  data <- .Call(rnng_request, context, data, send_mode, recv_mode, timeout, cv, msgid, environment())
+  data <- .Call(rnng_request, context, data, send_mode, recv_mode, timeout, cv, id, environment())

man/request.Rd

@@ -14,7 +14,6 @@ SEXP nano_HeadersSymbol;
 SEXP nano_IdSymbol;
 SEXP nano_ListenerSymbol;
 SEXP nano_MonitorSymbol;
-SEXP nano_MsgidSymbol;
 SEXP nano_ProtocolSymbol;
 SEXP nano_ResolveSymbol;
 SEXP nano_ResponseSymbol;
@@ -49,7 +48,6 @@ static void RegisterSymbols(void) {
   nano_IdSymbol = Rf_install("id");
   nano_ListenerSymbol = Rf_install("listener");
   nano_MonitorSymbol = Rf_install("monitor");
-  nano_MsgidSymbol = Rf_install("msgid");
   nano_ProtocolSymbol = Rf_install("protocol");
   nano_ResolveSymbol = Rf_install("resolve");
   nano_ResponseSymbol = Rf_install("response");

src/init.c

@@ -189,7 +189,7 @@ typedef struct nano_saio_s {
   nng_ctx *ctx;
   nng_aio *aio;
   void *cb;
-  int msgid;
+  int id;
 } nano_saio;
 
 typedef struct nano_cv_s {
@@ -255,7 +255,6 @@ extern SEXP nano_HeadersSymbol;
 extern SEXP nano_IdSymbol;
 extern SEXP nano_ListenerSymbol;
 extern SEXP nano_MonitorSymbol;
-extern SEXP nano_MsgidSymbol;
 extern SEXP nano_ProtocolSymbol;
 extern SEXP nano_ResolveSymbol;
 extern SEXP nano_ResponseSymbol;

src/nanonext.h

@@ -36,16 +36,13 @@ static void request_complete(void *arg) {
     raio->data = msg;
     nng_pipe p = nng_msg_get_pipe(msg);
     res = - (int) p.id;
-  } else if (res == 5) {
-    const int id = saio->msgid;
-    if (id) {
-      nng_msg *msg = NULL;
-      if (nng_msg_alloc(&msg, 0) == 0) {
-        if (nng_msg_append_u32(msg, 0) ||
-            nng_msg_append(msg, &id, sizeof(id)) ||
-            nng_ctx_sendmsg(*saio->ctx, msg, 0)) {
-          nng_msg_free(msg);
-        }
+  } else if (res == 5 && saio->id) {
+    nng_msg *msg = NULL;
+    if (nng_msg_alloc(&msg, 0) == 0) {
+      if (nng_msg_append_u32(msg, 0) ||
+          nng_msg_append(msg, &saio->id, sizeof(int)) ||
+          nng_ctx_sendmsg(*saio->ctx, msg, 0)) {
+        nng_msg_free(msg);
       }
     }
   }
@@ -447,7 +444,7 @@ SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeou
   nano_cv *ncv = signal ? (nano_cv *) NANO_PTR(cvar) : NULL;
 
   saio->ctx = ctx;
-  saio->msgid = id;
+  saio->id = id;
 
   if ((xc = nng_msg_alloc(&msg, 0)) ||
       (xc = nng_msg_append(msg, buf.buf, buf.cur)) ||
@@ -474,7 +471,7 @@ SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeou
   PROTECT(aio = R_MakeExternalPtr(raio, nano_AioSymbol, NANO_PROT(con)));
   R_RegisterCFinalizerEx(aio, request_finalizer, TRUE);
   Rf_setAttrib(aio, nano_ContextSymbol, con);
-  Rf_setAttrib(aio, nano_MsgidSymbol, Rf_ScalarInteger(id));
+  Rf_setAttrib(aio, nano_IdSymbol, Rf_ScalarInteger(id));
 
   PROTECT(env = R_NewEnv(R_NilValue, 0, 0));
   Rf_classgets(env, nano_reqAio);

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, msgid = 12L))
+test_class("recvAio", cs <- request(.context(req$socket), data = TRUE, timeout = 5, id = 12L))
 test_equal(5L, cs[])
 test_identical(c(0L, 12L), recv(ctxn, mode = 5L, block = 500))
 test_zero(reap(ctxn))
@@ -388,8 +388,6 @@ test_class("nanoSocket", poly1 <- socket(protocol = "poly"))
 test_class("nanoSocket", poly2 <- socket(protocol = "poly"))
 test_class("nanoMonitor", m <- monitor(poly, cv))
 test_print(m)
-test_class("sendAio", send_aio(poly, "one", timeout = 50))
-test_null(msleep(55))
 test_zero(listen(poly))
 test_null(read_monitor(m))
 test_zero(dial(poly1))