implement multi-threaded messenger

Author: shikokuchuo

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: nanonext
 Type: Package
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 0.3.0.9002
+Version: 0.3.0.9003
 Description: R binding for NNG (Nanomsg Next Gen), a successor to ZeroMQ. NNG is 
     a socket library providing high-performance scalability protocols,
     implementing a cross-platform standard for messaging and communications.

NAMESPACE

@@ -49,6 +49,7 @@ export(is_error_value)
 export(is_nul_byte)
 export(listen)
 export(logging)
+export(messenger)
 export(nano)
 export(ncurl)
 export(nng_error)

NEWS.md

@@ -1,9 +1,10 @@
-# nanonext 0.3.0.9002 (development)
+# nanonext 0.3.0.9003 (development)
 
 #### New Features
 
 * New `stream()` interface exposes low-level byte stream functionality in the NNG library, intended for communicating with non-NNG endpoints, including but not limited to websocket servers.
 * `ncurl()` adds an 'async' option to perform HTTP requests asynchronously, returning immediately with a 'recvAio'. Also adds explicit arguments for HTTP method, content type and authorization headers and request data.
+* New `messenger()` function implements a console-based 2-way messaging system using NNG's scalability protocols [currently experimental and undergoing early-stage testing].
 
 #### Updates
 

R/aio.R

@@ -1,4 +1,6 @@
-# nanonext - Core Functions - Aio Functions ------------------------------------
+# nanonext - Core - Aio Functions ----------------------------------------------
+
+# send_aio/recv_aio ------------------------------------------------------------
 
 #' Send Async
 #'
@@ -420,6 +422,8 @@ recv_aio.nanoStream <- function(con,
 
 }
 
+# Core aio functions -----------------------------------------------------------
+
 #' Call the Value of an Asynchronous AIO Operation
 #'
 #' Retrieve the value of an asynchronous AIO operation, waiting for the AIO

R/context.R

@@ -1,4 +1,4 @@
-# nanonext - Contexts ----------------------------------------------------------
+# nanonext - Contexts and RPC --------------------------------------------------
 
 #' Open Context
 #'
@@ -53,88 +53,6 @@ context <- function(socket) {
 
 }
 
-#' Send over Context
-#'
-#' Send data over a Context [Deprecated].
-#'
-#' @param context a Context.
-#' @inheritParams send
-#' @inheritParams send_aio
-#'
-#' @return Raw vector of sent data, or (invisibly) an integer exit code (zero on
-#'     success) if 'echo' is set to FALSE.
-#'
-#' @details Will block if the send is in progress and has not yet completed -
-#'     certain protocol / transport combinations may limit the number of messages
-#'     that can be queued if they have yet to be received. Set a timeout to
-#'     ensure the function returns under all scenarios.
-#'
-#' @keywords internal
-#' @export
-#'
-send_ctx <- function(context, data, mode = c("serial", "raw"), timeout, echo = TRUE) {
-
-  mode <- match.arg(mode)
-  if (missing(timeout)) timeout <- -2L
-  force(data)
-  data <- encode(data = data, mode = mode)
-  res <- .Call(rnng_ctx_send, context, data, timeout)
-  is.integer(res) && {
-    logerror(res)
-    return(invisible(res))
-  }
-  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
-
-}
-
-#' Receive over Context
-#'
-#' Receive data over a Context [Deprecated].
-#'
-#' @param context a Context.
-#' @inheritParams recv
-#' @inheritParams send_aio
-#'
-#' @return Named list of 2 elements: 'raw' containing the received raw vector
-#'     and 'data' containing the converted object, or else the converted object
-#'     if 'keep.raw' is set to FALSE.
-#'
-#' @details Will block while awaiting the receive operation to complete.
-#'     Set a timeout to ensure that the function returns under all scenarios.
-#'
-#'     In case of an error, an integer 'errorValue' is returned (to be
-#'     distiguishable from an integer message value). This can be verified using
-#'     \code{\link{is_error_value}}.
-#'
-#'     If the raw data was successfully received but an error occurred in
-#'     unserialisation or data conversion (for example if the incorrect mode was
-#'     specified), the received raw vector will always be returned to allow for
-#'     the data to be recovered.
-#'
-#' @keywords internal
-#' @export
-#'
-recv_ctx <- function(context,
-                     mode = c("serial", "character", "complex", "double",
-                              "integer", "logical", "numeric", "raw"),
-                     timeout,
-                     keep.raw = TRUE) {
-
-  mode <- match.arg(mode)
-  if (missing(timeout)) timeout <- -2L
-  res <- .Call(rnng_ctx_recv, context, timeout)
-  is.integer(res) && {
-    logerror(res)
-    return(invisible(`class<-`(res, "errorValue")))
-  }
-  on.exit(expr = return(res))
-  data <- decode(con = res, mode = mode)
-  on.exit()
-  missing(data) && return(.Call(rnng_scm))
-  if (missing(keep.raw) || isTRUE(keep.raw)) list(raw = res, data = data) else data
-
-}
-
 #' Reply over Context (RPC Server for Req/Rep Protocol)
 #'
 #' Implements an executor/server for the rep node of the req/rep protocol. Awaits
@@ -349,3 +267,87 @@ request <- function(context,
 
 }
 
+# Deprecated - do not use ------------------------------------------------------
+
+#' Send over Context
+#'
+#' Send data over a Context [Deprecated].
+#'
+#' @param context a Context.
+#' @inheritParams send
+#' @inheritParams send_aio
+#'
+#' @return Raw vector of sent data, or (invisibly) an integer exit code (zero on
+#'     success) if 'echo' is set to FALSE.
+#'
+#' @details Will block if the send is in progress and has not yet completed -
+#'     certain protocol / transport combinations may limit the number of messages
+#'     that can be queued if they have yet to be received. Set a timeout to
+#'     ensure the function returns under all scenarios.
+#'
+#' @keywords internal
+#' @export
+#'
+send_ctx <- function(context, data, mode = c("serial", "raw"), timeout, echo = TRUE) {
+
+  mode <- match.arg(mode)
+  if (missing(timeout)) timeout <- -2L
+  force(data)
+  data <- encode(data = data, mode = mode)
+  res <- .Call(rnng_ctx_send, context, data, timeout)
+  is.integer(res) && {
+    logerror(res)
+    return(invisible(res))
+  }
+  if (missing(echo) || isTRUE(echo)) res else invisible(0L)
+
+}
+
+#' Receive over Context
+#'
+#' Receive data over a Context [Deprecated].
+#'
+#' @param context a Context.
+#' @inheritParams recv
+#' @inheritParams send_aio
+#'
+#' @return Named list of 2 elements: 'raw' containing the received raw vector
+#'     and 'data' containing the converted object, or else the converted object
+#'     if 'keep.raw' is set to FALSE.
+#'
+#' @details Will block while awaiting the receive operation to complete.
+#'     Set a timeout to ensure that the function returns under all scenarios.
+#'
+#'     In case of an error, an integer 'errorValue' is returned (to be
+#'     distiguishable from an integer message value). This can be verified using
+#'     \code{\link{is_error_value}}.
+#'
+#'     If the raw data was successfully received but an error occurred in
+#'     unserialisation or data conversion (for example if the incorrect mode was
+#'     specified), the received raw vector will always be returned to allow for
+#'     the data to be recovered.
+#'
+#' @keywords internal
+#' @export
+#'
+recv_ctx <- function(context,
+                     mode = c("serial", "character", "complex", "double",
+                              "integer", "logical", "numeric", "raw"),
+                     timeout,
+                     keep.raw = TRUE) {
+
+  mode <- match.arg(mode)
+  if (missing(timeout)) timeout <- -2L
+  res <- .Call(rnng_ctx_recv, context, timeout)
+  is.integer(res) && {
+    logerror(res)
+    return(invisible(`class<-`(res, "errorValue")))
+  }
+  on.exit(expr = return(res))
+  data <- decode(con = res, mode = mode)
+  on.exit()
+  missing(data) && return(.Call(rnng_scm))
+  if (missing(keep.raw) || isTRUE(keep.raw)) list(raw = res, data = data) else data
+
+}
+

R/messenger.R

@@ -0,0 +1,66 @@
+# nanonext - Messenger ---------------------------------------------------------
+
+#' Messenger
+#'
+#' Console-based 2-way messaging system based on NNG scalability protocols.
+#'
+#' @param dial (optional) a URL to dial, 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.
+#'
+#' @section Usage:
+#'
+#'     Type outgoing messages and hit return to send.
+#'     Incoming messages are prefixed by \code{>}.
+#'
+#'     \code{:q} is the command to quit.
+#'
+#'     NOTE: This is an experimental feature that is currently undergoing
+#'     early-stage testing.
+#'
+#' @export
+#'
+messenger <- function(dial = NULL, listen = NULL) {
+
+  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.integer(sock) && {
+    logerror(sock)
+    return(invisible(sock))
+  }
+  on.exit(expr = {
+    close(sock)
+    invisible()
+  })
+  . <- unlist(strsplit("nanonext messenger", ""))
+  .. <- .[length(.):1]
+  for (i in seq_along(..)) {
+    cat("\r", `length<-`(.., i), sep = " ", file = stdout())
+    if (i %in% c(1:5, 15:20)) Sys.sleep(0.03) else Sys.sleep(0.01)
+  }
+  for (i in seq_along(.)) {
+    cat("\r", `length<-`(., i), sep = " ", file = stdout())
+    if (i %in% c(1:5, 15:20)) Sys.sleep(0.01) else Sys.sleep(0.03)
+  }
+  cat("  |  type your message:\n", file = stdout())
+  repeat {
+    data <- readline()
+    if (identical(data, ":q")) break
+    data <- writeBin(object = data, con = raw())
+    s <- .Call(rnng_send, sock, data, FALSE)
+    if (is.integer(s)) message(sprintf("%s [ no connection ] message not sent", format.POSIXct(Sys.time())))
+  }
+  on.exit()
+  close(sock)
+  invisible()
+
+}
+