Pipe event callback registration concept

shikokuchuo · shikokuchuo · commit fadcef5ad59b · 2025-04-30T11:38:19.000+01:00
diff --git a/NAMESPACE b/NAMESPACE
@@ -81,6 +81,7 @@ export(nng_version)
 export(opt)
 export(parse_url)
 export(pipe_notify)
+export(pipe_register)
 export(random)
 export(read_monitor)
 export(reap)
diff --git a/R/sync.R b/R/sync.R
@@ -155,6 +155,8 @@ cv_signal <- function(cv) invisible(.Call(rnng_cv_signal, cv))
 #' socket. The underlying transport may be closed at this point, and it is not
 #' possible to communicate using this pipe.
 #'
+#' Note: this function cannot be used in conjunction with [pipe_register()].
+#'
 #' @param socket a Socket.
 #' @param cv a 'conditionVariable' to signal, or NULL to cancel a previously set
 #'   signal.
@@ -197,6 +199,44 @@ cv_signal <- function(cv) invisible(.Call(rnng_cv_signal, cv))
 pipe_notify <- function(socket, cv, add = FALSE, remove = FALSE, flag = FALSE)
   invisible(.Call(rnng_pipe_notify, socket, cv, add, remove, flag))
 
+
+#' Pipe Register
+#'
+#' Register a callback to run whenever pipes (individual connections) are
+#' added or removed at a socket.
+#'
+#' For add: this event occurs after the pipe is fully added to the socket. Prior
+#' to this time, it is not possible to communicate over the pipe with the
+#' socket.
+#'
+#' For remove: this event occurs after the pipe has been removed from the
+#' socket. The underlying transport may be closed at this point, and it is not
+#' possible to communicate using this pipe.
+#'
+#' Note: this function cannot be used in conjunction with [pipe_notify()].
+#'
+#' @param socket a Socket.
+#' @param add \[default NULL\] an R function callback to be run whenever a pipe
+#'   is added. If NULL, any previously-registered callback is cancelled.
+#' @param remove \[default NULL\] an R function callback to be run whenever a
+#'   pipe is removed. If NULL, any previously-registered callback is cancelled.
+#'
+#' @return Invisibly, zero on success (will otherwise error).
+#'
+#' @examplesIf requireNamespace("later", quietly = TRUE)
+#' s <- socket(listen = "inproc://nanopipecb")
+#' pipe_register(s, function() print("hi"), function() print("bye"))
+#'
+#' s1 <- socket(dial = "inproc://nanopipecb")
+#' close(s1)
+#'
+#' close(s)
+#'
+#' @export
+#'
+pipe_register <- function(socket, add = NULL, remove = NULL)
+  invisible(.Call(rnng_pipe_register, socket, add, remove))
+
 #' Signal Forwarder
 #'
 #' Forwards signals from one 'conditionVariable' to another.
diff --git a/man/pipe_notify.Rd b/man/pipe_notify.Rd
diff --git a/man/pipe_register.Rd b/man/pipe_register.Rd
diff --git a/src/init.c b/src/init.c
@@ -149,6 +149,7 @@ static const R_CallMethodDef callMethods[] = {
   {"rnng_ncurl_session_close", (DL_FUNC) &rnng_ncurl_session_close, 1},
   {"rnng_ncurl_transact", (DL_FUNC) &rnng_ncurl_transact, 1},
   {"rnng_pipe_notify", (DL_FUNC) &rnng_pipe_notify, 5},
+  {"rnng_pipe_register", (DL_FUNC) &rnng_pipe_register, 3},
   {"rnng_protocol_open", (DL_FUNC) &rnng_protocol_open, 6},
   {"rnng_random", (DL_FUNC) &rnng_random, 2},
   {"rnng_reap", (DL_FUNC) &rnng_reap, 1},
diff --git a/src/nanonext.h b/src/nanonext.h
@@ -119,6 +119,7 @@ extern int R_interrupts_pending;
 
 #define ERROR_OUT(xc) Rf_error("%d | %s", xc, nng_strerror(xc))
 #define ERROR_RET(xc) { Rf_warning("%d | %s", xc, nng_strerror(xc)); return mk_error(xc); }
+#define NANONEXT_ENSURE_LATER if (eln2 == NULL) nano_load_later()
 #define NANONEXT_INIT_BUFSIZE 4096
 #define NANONEXT_SERIAL_VER 3
 #define NANONEXT_SERIAL_THR 134217728
@@ -345,6 +346,7 @@ SEXP rnng_ncurl_session(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_ncurl_session_close(SEXP);
 SEXP rnng_ncurl_transact(SEXP);
 SEXP rnng_pipe_notify(SEXP, SEXP, SEXP, SEXP, SEXP);
+SEXP rnng_pipe_register(SEXP, SEXP, SEXP);
 SEXP rnng_protocol_open(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_random(SEXP, SEXP);
 SEXP rnng_reap(SEXP);
diff --git a/src/sync.c b/src/sync.c
@@ -3,6 +3,29 @@
 #define NANONEXT_SIGNALS
 #include "nanonext.h"
 
+// internals -------------------------------------------------------------------
+
+static void nano_load_later(void) {
+
+  SEXP str, call;
+  PROTECT(str = Rf_mkString("later"));
+  PROTECT(call = Rf_lang2(Rf_install("loadNamespace"), str));
+  Rf_eval(call, R_BaseEnv);
+  UNPROTECT(2);
+  eln2 = (void (*)(void (*)(void *), void *, double, int)) R_GetCCallable("later", "execLaterNative2");
+
+}
+
+static void nano_eval_later (void *arg) {
+
+  SEXP call, node = (SEXP) arg;
+  node = R_WeakRefValue(node);
+  PROTECT(call = Rf_lcons(node, R_NilValue));
+  (void) Rf_eval(call, R_GlobalEnv);
+  UNPROTECT(1);
+
+}
+
 // aio completion callbacks ----------------------------------------------------
 
 static void sendaio_complete(void *arg) {
@@ -101,6 +124,12 @@ void pipe_cb_signal(nng_pipe p, nng_pipe_ev ev, void *arg) {
 
 }
 
+void pipe_cb_eval(nng_pipe p, nng_pipe_ev ev, void *arg) {
+
+  later2(nano_eval_later, arg);
+
+}
+
 static void pipe_cb_monitor(nng_pipe p, nng_pipe_ev ev, void *arg) {
 
   nano_monitor *monitor = (nano_monitor *) arg;
@@ -495,14 +524,7 @@ SEXP rnng_set_promise_context(SEXP x, SEXP ctx) {
 
   nano_aio *raio = (nano_aio *) NANO_PTR(aio);
 
-  if (eln2 == NULL) {
-    SEXP str, call;
-    PROTECT(str = Rf_mkString("later"));
-    PROTECT(call = Rf_lang2(Rf_install("loadNamespace"), str));
-    Rf_eval(call, R_BaseEnv);
-    UNPROTECT(2);
-    eln2 = (void (*)(void (*)(void *), void *, double, int)) R_GetCCallable("later", "execLaterNative2");
-  }
+  NANONEXT_ENSURE_LATER;
 
   switch (raio->type) {
   case REQAIO:
@@ -569,6 +591,52 @@ SEXP rnng_pipe_notify(SEXP socket, SEXP cv, SEXP add, SEXP remove, SEXP flag) {
 
 }
 
+SEXP rnng_pipe_register(SEXP socket, SEXP add, SEXP remove) {
+
+  if (NANO_PTR_CHECK(socket, nano_SocketSymbol))
+    Rf_error("'socket' is not a valid Socket");
+
+  int xc;
+  nng_socket *sock = (nng_socket *) NANO_PTR(socket);
+
+  NANONEXT_ENSURE_LATER;
+
+  if (add != R_NilValue) {
+
+    if (TYPEOF(add) != CLOSXP)
+      Rf_error("'add' is not a valid R closure function");
+
+    SEXP ref = R_MakeWeakRef(socket, add, R_NilValue, FALSE);
+    if ((xc = nng_pipe_notify(*sock, NNG_PIPE_EV_ADD_POST, pipe_cb_eval, ref)))
+      ERROR_OUT(xc);
+
+  } else {
+
+    if ((xc = nng_pipe_notify(*sock, NNG_PIPE_EV_ADD_POST, NULL, NULL)))
+      ERROR_OUT(xc);
+
+  }
+
+  if (remove != R_NilValue) {
+
+    if (TYPEOF(remove) != CLOSXP)
+      Rf_error("'remove' is not a valid R closure function");
+
+    SEXP ref = R_MakeWeakRef(socket, remove, R_NilValue, FALSE);
+    if ((xc = nng_pipe_notify(*sock, NNG_PIPE_EV_REM_POST, pipe_cb_eval, ref)))
+      ERROR_OUT(xc);
+
+  } else {
+
+    if ((xc = nng_pipe_notify(*sock, NNG_PIPE_EV_REM_POST, NULL, NULL)))
+      ERROR_OUT(xc);
+
+  }
+
+  return nano_success;
+
+}
+
 // monitors --------------------------------------------------------------------
 
 SEXP rnng_monitor_create(SEXP socket, SEXP cv) {
