consistent messenger interface

Author: shikokuchuo

NEWS.md

@@ -11,7 +11,7 @@
 * Unified `send()` and `recv()` functions, and their asynchronous counterparts `send_aio()` and `recv_aio()`, are now S3 generics and can be used across Sockets, Contexts and Streams.
 * Revised 'block' argument for `send()` and `recv()` now allows an integer value for setting a timeout.
 * `send_ctx()` and `recv_ctx()` are deprecated and will be removed in a future package version - the methods for `send()` and `recv()` should be used instead.
-* Logging is deprecated and is being phased out. Logging can still be set via 'NANONEXT_LOG' prior to package load but the ability to set via the `logging()` function has been removed. Logging will be removed entirely in a future package version.
+* Logging is deprecated and in the process of being phased out. Logging can still be set via 'NANONEXT_LOG' prior to package load but `logging()` can no longer be used. Logging will be removed entirely in a future package version.
 
 # nanonext 0.3.0
 

R/messenger.R

@@ -4,10 +4,8 @@
 #'
 #' Console-based 2-way messaging system based on NNG scalability protocols.
 #'
-#' @param dial (optional) a URL to dial, specifying the transport and address as
+#' @param url a URL to connect to, specifying the transport and address as
 #'     a character string e.g. 'tcp://127.0.0.1:5555' (see \link{transports}).
-#' @param listen (optional) a URL to listen at, specifying the transport and
-#'     address as a character string e.g. 'tcp://127.0.0.1:5555' (see \link{transports}).
 #'
 #' @return Invisible NULL.
 #'
@@ -23,21 +21,16 @@
 #'
 #' @export
 #'
-messenger <- function(dial = NULL, listen = NULL) {
+messenger <- function(url) {
 
-  if (!missing(dial) && is.character(dial)) {
-    sock <- .Call(rnng_messenger, dial, 1L)
-  } else if (!missing(listen) && is.character(listen)) {
-    sock <- .Call(rnng_messenger, listen, 0L)
-  } else {
-    stop("missing or invalid input")
-  }
+  is.character(url) || stop("the url must be supplied as a character string")
+  sock <- .Call(rnng_messenger, url)
   is.integer(sock) && {
     logerror(sock)
     return(invisible(sock))
   }
   on.exit(expr = {
-    s <- .Call(rnng_send, sock, as.raw(0L), 0L)
+    s <- .Call(rnng_send, sock, writeBin(":d ", raw()), 0L)
     close(sock)
     invisible()
   })
@@ -47,23 +40,28 @@ messenger <- function(dial = NULL, listen = NULL) {
     cat("\r", `length<-`(intro, i), sep = " ", file = stdout())
     Sys.sleep(0.02)
   }
-  cat("\n", file = stdout())
-  s <- .Call(rnng_send, sock, as.raw(0L), 0L)
+  cat(sprintf("\n| url: %s\n", url), file = stdout())
+  s <- .Call(rnng_send, sock, writeBin(":c ", raw()), 0L)
   if (is.integer(s)) {
-    cat(sprintf("| peer offline: waiting for connection: %s\n", format.POSIXct(Sys.time())),
+    cat(sprintf("| peer offline: %s\n", format.POSIXct(Sys.time())),
         file = stderr())
   } else {
-    cat(sprintf("| peer online: connected: %s\n", format.POSIXct(Sys.time())),
+    cat(sprintf("| peer online: %s\n", format.POSIXct(Sys.time())),
         file = stderr())
   }
   cat("type your message:\n", file = stdout())
   repeat {
     data <- readline()
     if (identical(data, ":q")) break
     if (identical(data, "")) next
-    data <- writeBin(object = data, con = raw())
-    s <- .Call(rnng_send, sock, data, 0L)
-    if (is.integer(s)) message(sprintf("| peer offline: message not sent > %s", format.POSIXct(Sys.time())))
+    rdata <- writeBin(object = data, con = raw())
+    s <- .Call(rnng_send, sock, rdata, 0L)
+    if (is.integer(s)) {
+      cat(sprintf("| peer offline: message not sent > %s\n", format.POSIXct(Sys.time())),
+          file = stderr())
+    } else {
+      cat(sprintf(" > %s\n", format.POSIXct(Sys.time())), file = stdout())
+    }
   }
 
 }

R/utils.R

@@ -88,6 +88,7 @@ is_error_value <- function(x) inherits(x, "errorValue")
 #'     package load. If the variable is set incorrectly, the default level
 #'     of 'error' is used instead.
 #'
+#' @keywords internal
 #' @export
 #'
 logging <- function(level = c("keep", "check", "error", "info")) {

man/logging.Rd

@@ -72,7 +72,7 @@ static const R_CallMethodDef CallEntries[] = {
   {"rnng_listener_set_string", (DL_FUNC) &rnng_listener_set_string, 3},
   {"rnng_listener_set_uint64", (DL_FUNC) &rnng_listener_set_uint64, 3},
   {"rnng_listener_start", (DL_FUNC) &rnng_listener_start, 1},
-  {"rnng_messenger", (DL_FUNC) &rnng_messenger, 2},
+  {"rnng_messenger", (DL_FUNC) &rnng_messenger, 1},
   {"rnng_ncurl", (DL_FUNC) &rnng_ncurl, 5},
   {"rnng_ncurl_aio", (DL_FUNC) &rnng_ncurl_aio, 5},
   {"rnng_protocol_open", (DL_FUNC) &rnng_protocol_open, 1},

man/messenger.Rd

@@ -63,7 +63,7 @@ extern SEXP rnng_listener_set_size(SEXP, SEXP, SEXP);
 extern SEXP rnng_listener_set_string(SEXP, SEXP, SEXP);
 extern SEXP rnng_listener_set_uint64(SEXP, SEXP, SEXP);
 extern SEXP rnng_listener_start(SEXP);
-extern SEXP rnng_messenger(SEXP, SEXP);
+extern SEXP rnng_messenger(SEXP);
 extern SEXP rnng_ncurl(SEXP, SEXP, SEXP, SEXP, SEXP);
 extern SEXP rnng_ncurl_aio(SEXP, SEXP, SEXP, SEXP, SEXP);
 extern SEXP rnng_protocol_open(SEXP);

src/init.c

@@ -98,7 +98,7 @@ static void thread_finalizer(SEXP xptr) {
 
 static void rnng_thread(void *arg) {
 
-  unsigned char *buf = NULL;
+  char *buf = NULL;
   size_t sz;
   int xc;
   time_t now;
@@ -110,19 +110,24 @@ static void rnng_thread(void *arg) {
     if (xc) {
       time(&now);
       tms = localtime(&now);
-      REprintf("| messenger session: ended: %d-%02d-%02d %02d:%02d:%02d\n",
+      REprintf("| messenger session ended: %d-%02d-%02d %02d:%02d:%02d\n",
                tms->tm_year + 1900, tms->tm_mon + 1, tms->tm_mday,
                tms->tm_hour, tms->tm_min, tms->tm_sec);
       break;
     }
-    if (!strcmp((const char *) buf, ":q")) {
+    if (!strcmp(buf, ":c ")) {
+      time(&now);
+      tms = localtime(&now);
+      REprintf("| peer connected: %d-%02d-%02d %02d:%02d:%02d\n",
+               tms->tm_year + 1900, tms->tm_mon + 1, tms->tm_mday,
+               tms->tm_hour, tms->tm_min, tms->tm_sec);
       nng_free(buf, sz);
-      break;
+      continue;
     }
-    if (!strcmp((const char *) buf, "")) {
+    if (!strcmp(buf, ":d ")) {
       time(&now);
       tms = localtime(&now);
-      REprintf("| peer status: changed: %d-%02d-%02d %02d:%02d:%02d\n",
+      REprintf("| peer disconnected: %d-%02d-%02d %02d:%02d:%02d\n",
                tms->tm_year + 1900, tms->tm_mon + 1, tms->tm_mday,
                tms->tm_hour, tms->tm_min, tms->tm_sec);
       nng_free(buf, sz);
@@ -131,7 +136,7 @@ static void rnng_thread(void *arg) {
 
     time(&now);
     tms = localtime(&now);
-    Rprintf("%s < %d-%02d-%02d %02d:%02d:%02d\n",
+    Rprintf("%s\n < %d-%02d-%02d %02d:%02d:%02d\n",
             buf,
             tms->tm_year + 1900, tms->tm_mon + 1, tms->tm_mday,
             tms->tm_hour, tms->tm_min, tms->tm_sec);
@@ -140,51 +145,44 @@ static void rnng_thread(void *arg) {
 
 }
 
-SEXP rnng_messenger(SEXP url, SEXP mode) {
+SEXP rnng_messenger(SEXP url) {
 
   const char *up = CHAR(STRING_ELT(url, 0));
-  int dial = INTEGER(mode)[0];
-
   SEXP con;
   SEXP socket;
   int xc;
+  nng_listener *dlp;
 
   nng_socket *sock = R_Calloc(1, nng_socket);
   xc = nng_pair0_open(sock);
   if (xc) {
     R_Free(sock);
     return Rf_ScalarInteger(xc);
   }
-
-  if (dial) {
-    nng_dialer *dp = R_Calloc(1, nng_dialer);
-    int xc = nng_dial(*sock, up, dp, 2u);
+  dlp = R_Calloc(1, nng_listener);
+  xc = nng_listen(*sock, up, dlp, 0);
+  if (xc == 10) {
+    R_Free(dlp);
+    nng_dialer *dlp = R_Calloc(1, nng_dialer);
+    xc = nng_dial(*sock, up, dlp, 2u);
     if (xc) {
-      R_Free(dp);
+      R_Free(dlp);
       R_Free(sock);
       return Rf_ScalarInteger(xc);
     }
-    socket = PROTECT(R_MakeExternalPtr(sock, nano_SocketSymbol, R_NilValue));
-    R_RegisterCFinalizerEx(socket, socket_finalizer, TRUE);
-    con = PROTECT(R_MakeExternalPtr(dp, nano_DialerSymbol, R_NilValue));
-    R_MakeWeakRef(socket, con, R_NilValue, TRUE);
-    UNPROTECT(1);
-
-  } else {
-    nng_listener *lp = R_Calloc(1, nng_listener);
-    int xc = nng_listen(*sock, up, lp, 0);
-    if (xc) {
-      R_Free(lp);
-      R_Free(sock);
-      return Rf_ScalarInteger(xc);
-    }
-    socket = PROTECT(R_MakeExternalPtr(sock, nano_SocketSymbol, R_NilValue));
-    R_RegisterCFinalizerEx(socket, socket_finalizer, TRUE);
-    con = PROTECT(R_MakeExternalPtr(lp, nano_ListenerSymbol, R_NilValue));
-    R_MakeWeakRef(socket, con, R_NilValue, TRUE);
-    UNPROTECT(1);
+
+  } else if (xc) {
+    R_Free(dlp);
+    R_Free(sock);
+    return Rf_ScalarInteger(xc);
   }
 
+  socket = PROTECT(R_MakeExternalPtr(sock, nano_SocketSymbol, R_NilValue));
+  R_RegisterCFinalizerEx(socket, socket_finalizer, TRUE);
+  con = PROTECT(R_MakeExternalPtr(dlp, nano_ListenerSymbol, R_NilValue));
+  R_MakeWeakRef(socket, con, R_NilValue, TRUE);
+  UNPROTECT(1);
+
   nng_thread *thr;
   nng_thread_create(&thr, rnng_thread, sock);
   SEXP xptr = PROTECT(R_MakeExternalPtr(thr, R_NilValue, R_NilValue));

src/nanonext.h

@@ -99,7 +99,6 @@ invisible(close(sock2) == 0L || stop())
 
 invisible(is.integer(suppressMessages(ncurl("http://127.0.0.1:5555"))) || stop())
 invisible(is.integer(suppressMessages(stream(dial = "tcp://127.0.0.1:5555"))) || stop())
-logging(level = "keep")
 
 invisible(is.character(ver <- nng_version()) && length(ver) == 2L || stop())
 invisible(is.character(nng_error(0L)) || stop())