re-restrict request/reply to contexts

Author: shikokuchuo

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: nanonext
 Type: Package
 Title: NNG (Nanomsg Next Gen) Lightweight Messaging Library
-Version: 0.13.3.9003
+Version: 0.13.3.9004
 Description: R binding for NNG (Nanomsg Next Gen), a successor to ZeroMQ. NNG is
     a socket library implementing 'Scalability Protocols', a reliable,
     high-performance standard for common communications patterns including

NEWS.md

@@ -1,6 +1,5 @@
-# nanonext 0.13.3.9003 (development)
+# nanonext 0.13.3.9004 (development)
 
-* `request()` and `reply()` now accept sockets in place of contexts, with the argument 'context' amended to the more general 'con'.
 * An integer file descriptor is appended to 'nanoSockets' as the attribute 'fd' - see updated documentation for `socket()`.
 
 # nanonext 0.13.3

R/context.R

@@ -91,13 +91,13 @@ context <- function(socket) .Call(rnng_ctx_open, socket)
 #'
 close.nanoContext <- function(con, ...) invisible(.Call(rnng_ctx_close, con))
 
-#' Reply (RPC Server for Req/Rep Protocol)
+#' Reply over Context (RPC Server for Req/Rep Protocol)
 #'
 #' Implements an executor/server for the rep node of the req/rep protocol.
 #'     Awaits data, applies an arbitrary specified function, and returns the
 #'     result to the caller/client.
 #'
-#' @param con a Socket or Context.
+#' @param context a Context.
 #' @param execute a function which takes the received (converted) data as its
 #'     first argument. Can be an anonymous function of the form
 #'     \code{function(x) do(x)}. Additional arguments can also be passed in
@@ -154,7 +154,7 @@ close.nanoContext <- function(con, ...) invisible(.Call(rnng_ctx_close, con))
 #'
 #' @export
 #'
-reply <- function(con,
+reply <- function(context,
                   execute,
                   recv_mode = c("serial", "character", "complex", "double",
                                 "integer", "logical", "numeric", "raw"),
@@ -163,16 +163,16 @@ reply <- function(con,
                   ...) {
 
   block <- if (is.null(timeout)) TRUE else timeout
-  res <- recv(con, mode = recv_mode, block = block)
+  res <- recv(context, mode = recv_mode, block = block)
   is_error_value(res) && return(res)
-  on.exit(expr = send(con, data = as.raw(0L), mode = send_mode, block = TRUE))
+  on.exit(expr = send(context, data = as.raw(0L), mode = send_mode, block = TRUE))
   data <- execute(res, ...)
   on.exit()
-  send(con, data = data, mode = send_mode, block = block)
+  send(context, data = data, mode = send_mode, block = block)
 
 }
 
-#' Request (RPC Client for Req/Rep Protocol)
+#' Request over Context (RPC Client for Req/Rep Protocol)
 #'
 #' Implements a caller/client for the req node of the req/rep protocol. Sends
 #'     data to the rep node (executor/server) and returns an Aio, which can be
@@ -222,13 +222,13 @@ reply <- function(con,
 #'
 #' @export
 #'
-request <- function(con,
+request <- function(context,
                     data,
                     send_mode = c("serial", "raw", "next"),
                     recv_mode = c("serial", "character", "complex", "double",
                                   "integer", "logical", "numeric", "raw", "string"),
                     timeout = NULL)
-  data <- .Call(rnng_request, con, data, send_mode, recv_mode, timeout, environment())
+  data <- .Call(rnng_request, context, data, send_mode, recv_mode, timeout, environment())
 
 #' Request and Signal a Condition Variable (RPC Client for Req/Rep Protocol)
 #'
@@ -261,11 +261,11 @@ request <- function(con,
 #' @rdname request
 #' @export
 #'
-request_signal <- function(con,
+request_signal <- function(context,
                            data,
                            cv,
                            send_mode = c("serial", "raw", "next"),
                            recv_mode = c("serial", "character", "complex", "double",
                                          "integer", "logical", "numeric", "raw", "string"),
                            timeout = NULL)
-  data <- .Call(rnng_request_signal, con, data, cv, send_mode, recv_mode, timeout, environment())
+  data <- .Call(rnng_request_signal, context, data, cv, send_mode, recv_mode, timeout, environment())

man/reply.Rd

@@ -322,20 +322,6 @@ static void request_finalizer(SEXP xptr) {
 
 }
 
-static void request_sock_finalizer(SEXP xptr) {
-
-  if (R_ExternalPtrAddr(xptr) == NULL) return;
-  nano_aio *xp = (nano_aio *) R_ExternalPtrAddr(xptr);
-  nano_aio *saio = (nano_aio *) xp->next;
-  nng_aio_free(saio->aio);
-  nng_aio_free(xp->aio);
-  if (xp->data != NULL)
-    nng_msg_free((nng_msg *) xp->data);
-  R_Free(saio);
-  R_Free(xp);
-
-}
-
 static void cv_finalizer(SEXP xptr) {
 
   if (R_ExternalPtrAddr(xptr) == NULL) return;
@@ -1225,84 +1211,9 @@ SEXP rnng_ncurl_session_close(SEXP session) {
 
 // request ---------------------------------------------------------------------
 
-SEXP rnng_request_sock(const SEXP con, const SEXP data, const SEXP sendmode,
+SEXP rnng_request_impl(const SEXP con, const SEXP data, const SEXP sendmode,
                        const SEXP recvmode, const SEXP timeout, const SEXP clo, nano_cv *ncv) {
 
-  const nng_duration dur = timeout == R_NilValue ? NNG_DURATION_DEFAULT : (nng_duration) Rf_asInteger(timeout);
-  const int mod = nano_matcharg(recvmode);
-  const int signal = ncv != NULL;
-  nng_socket *sock = (nng_socket *) R_ExternalPtrAddr(con);
-  SEXP aio, env, fun;
-  nano_buf buf;
-  nano_aio *saio, *raio;
-  nng_msg *msg;
-  int xc;
-
-  switch (nano_encodes(sendmode)) {
-  case 1:
-    nano_serialize(&buf, data); break;
-  case 2:
-    nano_encode(&buf, data); break;
-  default:
-    nano_serialize_next(&buf, data); break;
-  }
-
-  saio = R_Calloc(1, nano_aio);
-  saio->next = ncv;
-
-  if ((xc = nng_msg_alloc(&msg, 0)))
-    goto exitlevel1;
-
-  if ((xc = nng_msg_append(msg, buf.buf, buf.cur)) ||
-      (xc = nng_aio_alloc(&saio->aio, sendaio_complete, &saio->aio))) {
-    nng_msg_free(msg);
-    goto exitlevel1;
-  }
-
-  nng_aio_set_msg(saio->aio, msg);
-  nng_send_aio(*sock, saio->aio);
-
-  raio = R_Calloc(1, nano_aio);
-  raio->type = RECVAIO;
-  raio->mode = mod;
-  raio->next = saio;
-
-  if ((xc = nng_aio_alloc(&raio->aio, signal ? request_complete_signal : raio_complete, raio)))
-    goto exitlevel2;
-
-  nng_aio_set_timeout(raio->aio, dur);
-  nng_recv_aio(*sock, raio->aio);
-  NANO_FREE(buf);
-
-  PROTECT(aio = R_MakeExternalPtr(raio, nano_AioSymbol, R_NilValue));
-  R_RegisterCFinalizerEx(aio, request_sock_finalizer, TRUE);
-
-  PROTECT(env = Rf_allocSExp(ENVSXP));
-  NANO_CLASS(env, "recvAio");
-  Rf_defineVar(nano_AioSymbol, aio, env);
-
-  PROTECT(fun = Rf_allocSExp(CLOSXP));
-  SET_FORMALS(fun, nano_aioFormals);
-  SET_BODY(fun, signal ? CADDDR(nano_aioFuncs) : CADR(nano_aioFuncs));
-  SET_CLOENV(fun, clo);
-  R_MakeActiveBinding(nano_DataSymbol, fun, env);
-
-  UNPROTECT(3);
-  return env;
-
-  exitlevel2:
-  R_Free(raio);
-  nng_aio_free(saio->aio);
-  exitlevel1:
-  R_Free(saio);
-  NANO_FREE(buf);
-  return mk_error_data(xc);
-
-}
-
-SEXP rnng_request_ctx(const SEXP con, const SEXP data, const SEXP sendmode,
-                      const SEXP recvmode, const SEXP timeout, const SEXP clo, nano_cv *ncv) {
-
   const nng_duration dur = timeout == R_NilValue ? NNG_DURATION_DEFAULT : (nng_duration) Rf_asInteger(timeout);
   const int mod = nano_matcharg(recvmode);
   const int signal = ncv != NULL;
@@ -1380,31 +1291,22 @@ SEXP rnng_request_ctx(const SEXP con, const SEXP data, const SEXP sendmode,
 
 SEXP rnng_request(SEXP con, SEXP data, SEXP sendmode, SEXP recvmode, SEXP timeout, SEXP clo) {
 
-  const SEXP ptrtag = R_ExternalPtrTag(con);
-  if (ptrtag == nano_ContextSymbol) {
-    return rnng_request_ctx(con, data, sendmode, recvmode, timeout, clo, NULL);
-  } else if (ptrtag == nano_SocketSymbol) {
-    return rnng_request_sock(con, data, sendmode, recvmode, timeout, clo, NULL);
-  } else {
-    error_return("'con' is not a valid Socket or Context");
-  }
+  if (R_ExternalPtrTag(con) != nano_ContextSymbol)
+    Rf_error("'con' is not a valid Context");
+
+  return rnng_request_impl(con, data, sendmode, recvmode, timeout, clo, NULL);
 
 }
 
 SEXP rnng_request_signal(SEXP con, SEXP data, SEXP cvar, SEXP sendmode, SEXP recvmode, SEXP timeout, SEXP clo) {
 
+  if (R_ExternalPtrTag(con) != nano_ContextSymbol)
+    Rf_error("'con' is not a valid Context");
   if (R_ExternalPtrTag(cvar) != nano_CvSymbol)
     Rf_error("'cv' is not a valid Condition Variable");
   nano_cv *ncv = (nano_cv *) R_ExternalPtrAddr(cvar);
 
-  const SEXP ptrtag = R_ExternalPtrTag(con);
-  if (ptrtag == nano_ContextSymbol) {
-    return rnng_request_ctx(con, data, sendmode, recvmode, timeout, clo, ncv);
-  } else if (ptrtag == nano_SocketSymbol) {
-    return rnng_request_sock(con, data, sendmode, recvmode, timeout, clo, ncv);
-  } else {
-    error_return("'con' is not a valid Socket or Context");
-  }
+  return rnng_request_impl(con, data, sendmode, recvmode, timeout, clo, ncv);
 
 }
 

man/request.Rd

@@ -486,7 +486,7 @@ nanotest(!.unresolved(fakesock))
 fakectx <- `class<-`("test", "nanoContext")
 nanotest(!unresolved(fakectx))
 nanotest(!.unresolved(fakectx))
-nanotesterr(request(fakectx, data = "test"), "valid Socket or Context")
+nanotesterr(request(fakectx, data = "test"), "valid Context")
 nanotesterr(subscribe(fakectx, NULL), "valid")
 nanotesterr(close(fakectx), "valid Context")
 nanotest(reap(fakectx) == 3L)