more robust code

Author: shikokuchuo

.github/workflows/test-coverage.yaml

@@ -2,9 +2,9 @@
 # Need help debugging build failures? Start at https://github.com/r-lib/actions#where-to-find-help
 on:
   push:
-    branches: [main, master]
+    branches: [main]
   pull_request:
-    branches: [main, master]
+    branches: [main]
 
 name: test-coverage
 

R/sendrecv.R

@@ -222,7 +222,6 @@ recv.nanoSocket <- function(con,
 
   mode <- match.arg2(mode, c("serial", "character", "complex", "double",
                              "integer", "logical", "numeric", "raw"))
-  keep.raw <- missing(keep.raw) || isTRUE(keep.raw)
   .Call(rnng_recv, con, mode, block, keep.raw)
 
 }
@@ -241,7 +240,6 @@ recv.nanoContext <- function(con,
   mode <- match.arg2(mode, c("serial", "character", "complex", "double",
                              "integer", "logical", "numeric", "raw"))
   if (missing(block) || isTRUE(block)) block <- -2L
-  keep.raw <- missing(keep.raw) || isTRUE(keep.raw)
   .Call(rnng_ctx_recv, con, mode, block, keep.raw)
 
 }
@@ -261,7 +259,6 @@ recv.nanoStream <- function(con,
   mode <- match.arg2(mode, c("character", "complex", "double", "integer",
                              "logical", "numeric", "raw")) + 1L
   if (missing(block) || isTRUE(block)) block <- -2L
-  keep.raw <- missing(keep.raw) || isTRUE(keep.raw)
   .Call(rnng_stream_recv, con, mode, block, keep.raw, n)
 
 }

README.Rmd

@@ -437,9 +437,9 @@ In this respect, it may be used as a performant and lightweight method for makin
 
 ```{r stream}
 
-s <- stream(dial = "wss://socketsbay.com/wss/v2/2/demo/", textframes = TRUE)
+s <- stream(dial = "wss://demo.piesocket.com/v3/channel_1", textframes = TRUE)
 s
-s |> send("hello world")
+s |> recv(keep.raw = FALSE)
 
 ```
 

README.md

@@ -379,7 +379,7 @@ aio
 #> < recvAio >
 #>  - $data for message data
 aio$data |> str()
-#>  num [1:100000000] -0.597 -0.351 -0.984 1.93 1.393 ...
+#>  num [1:100000000] -0.0183 0.9349 -0.8196 0.0607 -0.1335 ...
 ```
 
 As `call_aio()` is blocking and will wait for completion, an alternative
@@ -516,11 +516,11 @@ ncurl("http://httpbin.org/headers")
 #>   [1] 7b 0a 20 20 22 68 65 61 64 65 72 73 22 3a 20 7b 0a 20 20 20 20 22 48 6f 73
 #>  [26] 74 22 3a 20 22 68 74 74 70 62 69 6e 2e 6f 72 67 22 2c 20 0a 20 20 20 20 22
 #>  [51] 58 2d 41 6d 7a 6e 2d 54 72 61 63 65 2d 49 64 22 3a 20 22 52 6f 6f 74 3d 31
-#>  [76] 2d 36 32 36 35 31 34 37 38 2d 31 38 64 62 65 31 36 38 34 32 62 63 62 36 62
-#> [101] 63 33 30 36 61 37 61 63 33 22 0a 20 20 7d 0a 7d 0a
+#>  [76] 2d 36 32 36 35 39 30 34 32 2d 34 38 34 35 64 63 30 36 36 32 66 30 63 38 66
+#> [101] 30 32 31 33 63 35 63 63 32 22 0a 20 20 7d 0a 7d 0a
 #> 
 #> $data
-#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62651478-18dbe16842bcb6bc306a7ac3\"\n  }\n}\n"
+#> [1] "{\n  \"headers\": {\n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62659042-4845dc0662f0c8f0213c5cc2\"\n  }\n}\n"
 ```
 
 For advanced use, supports additional HTTP methods such as POST or PUT.
@@ -535,7 +535,7 @@ res
 #>  - $raw for raw message
 
 call_aio(res)$data
-#> [1] "{\n  \"args\": {}, \n  \"data\": \"{\\\"key\\\": \\\"value\\\"}\", \n  \"files\": {}, \n  \"form\": {}, \n  \"headers\": {\n    \"Authorization\": \"Bearer APIKEY\", \n    \"Content-Length\": \"16\", \n    \"Content-Type\": \"application/json\", \n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62651478-224f907c244b4b5053cfdac3\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"78.145.225.121\", \n  \"url\": \"http://httpbin.org/post\"\n}\n"
+#> [1] "{\n  \"args\": {}, \n  \"data\": \"{\\\"key\\\": \\\"value\\\"}\", \n  \"files\": {}, \n  \"form\": {}, \n  \"headers\": {\n    \"Authorization\": \"Bearer APIKEY\", \n    \"Content-Length\": \"16\", \n    \"Content-Type\": \"application/json\", \n    \"Host\": \"httpbin.org\", \n    \"X-Amzn-Trace-Id\": \"Root=1-62659042-7ab689205d5bd45b62c030da\"\n  }, \n  \"json\": {\n    \"key\": \"value\"\n  }, \n  \"origin\": \"78.145.225.121\", \n  \"url\": \"http://httpbin.org/post\"\n}\n"
 ```
 
 In this respect, it may be used as a performant and lightweight method
@@ -550,14 +550,14 @@ communicating with raw sockets. This may be used for connecting to
 arbitrary non-NNG endpoints.
 
 ``` r
-s <- stream(dial = "wss://socketsbay.com/wss/v2/2/demo/", textframes = TRUE)
+s <- stream(dial = "wss://demo.piesocket.com/v3/channel_1", textframes = TRUE)
 s
 #> < nanoStream >
 #>  - type: dialer 
-#>  - url: wss://socketsbay.com/wss/v2/2/demo/ 
+#>  - url: wss://demo.piesocket.com/v3/channel_1 
 #>  - textframes: TRUE
-s |> send("hello world")
-#>  [1] 68 65 6c 6c 6f 20 77 6f 72 6c 64 00
+s |> recv(keep.raw = FALSE)
+#> [1] "{\"error\":\"Missing apiKey\"}"
 ```
 
 The stream interface can be used to communicate with websocket servers.

src/aio.c

@@ -181,7 +181,7 @@ SEXP rnng_aio_get_msg(SEXP aio, SEXP mode, SEXP keep) {
     return mk_error(res);
 
   const int mod = *INTEGER(mode), kpr = *LOGICAL(keep);
-  unsigned char *buf = nng_msg_body(raio->data);
+  unsigned char *buf = (unsigned char *) nng_msg_body(raio->data);
   size_t sz = nng_msg_len(raio->data);
 
   return nano_decode(buf, sz, mod, kpr);
@@ -210,7 +210,7 @@ SEXP rnng_aio_stream_in(SEXP aio, SEXP mode, SEXP keep) {
 
   const int mod = *INTEGER(mode), kpr = *LOGICAL(keep);
   nng_iov *iov = (nng_iov *) iaio->data;
-  unsigned char *buf = iov->iov_buf;
+  unsigned char *buf = (unsigned char *) iov->iov_buf;
   size_t sz = nng_aio_count(iaio->aio);
 
   return nano_decode(buf, sz, mod, kpr);
@@ -390,7 +390,7 @@ SEXP rnng_stream_recv_aio(SEXP stream, SEXP bytes, SEXP timeout) {
 
   nng_stream *sp = (nng_stream *) R_ExternalPtrAddr(stream);
   const nng_duration dur = (nng_duration) Rf_asInteger(timeout);
-  const size_t xlen = Rf_asInteger(bytes);
+  const size_t xlen = (size_t) Rf_asInteger(bytes);
   nano_aio *iaio = R_Calloc(1, nano_aio);
   nng_iov *iov = R_Calloc(1, nng_iov);
   int xc;
@@ -399,7 +399,7 @@ SEXP rnng_stream_recv_aio(SEXP stream, SEXP bytes, SEXP timeout) {
   iaio->type = IOV_RECVAIO;
   iaio->data = iov;
   iov->iov_len = xlen;
-  iov->iov_buf = R_Calloc(xlen, unsigned char);
+  iov->iov_buf = (unsigned char *) R_Calloc(xlen, unsigned char);
 
   xc = nng_aio_alloc(&iaio->aio, iaio_complete, iaio);
   if (xc) {
@@ -416,7 +416,7 @@ SEXP rnng_stream_recv_aio(SEXP stream, SEXP bytes, SEXP timeout) {
     R_Free(iaio);
     return mk_error(xc);
   }
-  xc = nng_aio_set_iov(iaio->aio, 1, iaio->data);
+  xc = nng_aio_set_iov(iaio->aio, 1u, iov);
   if (xc) {
     nng_mtx_free(iaio->mtx);
     nng_aio_free(iaio->aio);
@@ -579,7 +579,7 @@ SEXP rnng_stream_send_aio(SEXP stream, SEXP data, SEXP timeout) {
     R_Free(iaio);
     return Rf_ScalarInteger(xc);
   }
-  xc = nng_aio_set_iov(iaio->aio, 1, iaio->data);
+  xc = nng_aio_set_iov(iaio->aio, 1u, iov);
   if (xc) {
     nng_mtx_free(iaio->mtx);
     nng_aio_free(iaio->aio);
@@ -690,7 +690,7 @@ SEXP rnng_ncurl_aio(SEXP http, SEXP method, SEXP headers, SEXP data) {
   }
   if (data != R_NilValue) {
     unsigned char *dp = RAW(data);
-    const R_xlen_t dlen = Rf_xlength(data) - 1;
+    const size_t dlen = (size_t) Rf_xlength(data) - 1;
     xc = nng_http_req_set_data(handle->req, dp, dlen);
     if (xc) {
       nng_http_req_free(handle->req);
@@ -801,16 +801,13 @@ SEXP rnng_aio_http(SEXP aio) {
 
   void *dat;
   size_t sz;
-  unsigned char *rp;
   SEXP vec;
 
   nng_http_res_get_data(handle->res, &dat, &sz);
 
-  PROTECT(vec = Rf_allocVector(RAWSXP, sz));
-  rp = RAW(vec);
-  memcpy(rp, dat, sz);
+  vec = Rf_allocVector(RAWSXP, sz);
+  memcpy(RAW(vec), (unsigned char *) dat, sz);
 
-  UNPROTECT(1);
   return vec;
 
 }

src/core.c

@@ -20,14 +20,14 @@ SEXP mk_error(const int xc) {
 SEXP nano_encode(SEXP object) {
 
   R_xlen_t xlen = Rf_xlength(object);
+  unsigned char *buf;
   size_t sz;
   SEXP out;
 
   if (!Rf_isVectorAtomic(object))
     error_return("'data' is not an atomic vector type");
   if (TYPEOF(object) == STRSXP) {
     const char *s;
-    unsigned char *buf;
     size_t np, outlen = 0;
     R_xlen_t i;
     for (i = 0; i < xlen; i++)
@@ -43,24 +43,28 @@ SEXP nano_encode(SEXP object) {
   } else {
     switch (TYPEOF(object)) {
     case REALSXP:
+      buf = (unsigned char *) REAL(object);
       sz = xlen * sizeof(double);
       out = Rf_allocVector(RAWSXP, sz);
-      memcpy(RAW(out), REAL(object), sz);
+      memcpy(RAW(out), buf, sz);
       break;
     case INTSXP:
+      buf = (unsigned char *) INTEGER(object);
       sz = xlen * sizeof(int);
       out = Rf_allocVector(RAWSXP, sz);
-      memcpy(RAW(out), INTEGER(object), sz);
+      memcpy(RAW(out), buf, sz);
       break;
     case LGLSXP:
+      buf = (unsigned char *) LOGICAL(object);
       sz = xlen * sizeof(int);
       out = Rf_allocVector(RAWSXP, sz);
-      memcpy(RAW(out), LOGICAL(object), sz);
+      memcpy(RAW(out), buf, sz);
       break;
     case CPLXSXP:
+      buf = (unsigned char *) COMPLEX(object);
       sz = xlen * (sizeof(double) + sizeof(double));
       out = Rf_allocVector(RAWSXP, sz);
-      memcpy(RAW(out), COMPLEX(object), sz);
+      memcpy(RAW(out), buf, sz);
       break;
     case RAWSXP:
       out = object;
@@ -78,7 +82,7 @@ SEXP rawOneString(unsigned char *bytes, R_xlen_t nbytes, R_xlen_t *np) {
 
   unsigned char *p;
   R_xlen_t i;
-  char *buf;
+  char *cbuf;
   SEXP res;
 
   for (i = *np, p = bytes + (*np); i < nbytes; p++, i++)
@@ -89,11 +93,11 @@ SEXP rawOneString(unsigned char *bytes, R_xlen_t nbytes, R_xlen_t *np) {
     *np = i + 1;
     res = Rf_mkChar((char *) p);
   } else {
-    buf = R_chk_calloc(nbytes - (*np) + 1, 1);
-    memcpy(buf, bytes + (*np), nbytes - (*np));
+    cbuf = R_chk_calloc(nbytes - (*np) + 1, 1);
+    memcpy(cbuf, bytes + (*np), nbytes - (*np));
     *np = nbytes;
-    res = Rf_mkChar(buf);
-    R_Free(buf);
+    res = Rf_mkChar(cbuf);
+    R_Free(cbuf);
   }
 
   return res;
@@ -646,7 +650,9 @@ SEXP rnng_recv(SEXP socket, SEXP mode, SEXP block, SEXP keep) {
       return mk_error(xc);
     }
     nng_msg *msgp = nng_aio_get_msg(aiop);
-    res = nano_decode(nng_msg_body(msgp), nng_msg_len(msgp), mod, kpr);
+    buf = (unsigned char *) nng_msg_body(msgp);
+    sz = nng_msg_len(msgp);
+    res = nano_decode(buf, sz, mod, kpr);
     nng_msg_free(msgp);
     nng_aio_free(aiop);
   }
@@ -701,6 +707,8 @@ SEXP rnng_ctx_recv(SEXP context, SEXP mode, SEXP timeout, SEXP keep) {
   const nng_duration dur = (nng_duration) Rf_asInteger(timeout);
   const int mod = *INTEGER(mode), kpr = *LOGICAL(keep);
   int xc;
+  unsigned char *buf;
+  size_t sz;
   SEXP res;
 
   xc = nng_aio_alloc(&aiop, NULL, NULL);
@@ -717,7 +725,9 @@ SEXP rnng_ctx_recv(SEXP context, SEXP mode, SEXP timeout, SEXP keep) {
   }
 
   nng_msg *msgp = nng_aio_get_msg(aiop);
-  res = nano_decode(nng_msg_body(msgp), nng_msg_len(msgp), mod, kpr);
+  buf = (unsigned char *) nng_msg_body(msgp);
+  sz = nng_msg_len(msgp);
+  res = nano_decode(buf, sz, mod, kpr);
   nng_msg_free(msgp);
   nng_aio_free(aiop);
 
@@ -783,7 +793,7 @@ SEXP rnng_stream_recv(SEXP stream, SEXP mode, SEXP timeout, SEXP keep, SEXP byte
   SEXP res;
 
   iov.iov_len = xlen;
-  iov.iov_buf = R_Calloc(xlen, unsigned char);
+  iov.iov_buf = (unsigned char *) R_Calloc(xlen, unsigned char);
 
   xc = nng_aio_alloc(&aiop, NULL, NULL);
   if (xc) {

src/utils.c

@@ -154,7 +154,7 @@ SEXP rnng_ncurl(SEXP http, SEXP method, SEXP headers, SEXP data) {
   }
   if (data != R_NilValue) {
     unsigned char *dp = RAW(data);
-    const R_xlen_t dlen = XLENGTH(data) - 1;
+    const size_t dlen = (size_t) Rf_xlength(data) - 1;
     xc = nng_http_req_set_data(req, dp, dlen);
     if (xc) {
       nng_http_req_free(req);
@@ -234,22 +234,19 @@ SEXP rnng_ncurl(SEXP http, SEXP method, SEXP headers, SEXP data) {
 
   void *dat;
   size_t sz;
-  unsigned char *rp;
   SEXP vec;
 
   nng_http_res_get_data(res, &dat, &sz);
 
-  PROTECT(vec = Rf_allocVector(RAWSXP, sz));
-  rp = RAW(vec);
-  memcpy(rp, dat, sz);
+  vec = Rf_allocVector(RAWSXP, sz);
+  memcpy(RAW(vec), (unsigned char *) dat, sz);
   if (cfg != NULL)
     nng_tls_config_free(cfg);
   nng_http_res_free(res);
   nng_http_req_free(req);
   nng_http_client_free(client);
   nng_url_free(url);
 
-  UNPROTECT(1);
   return vec;
 
 }
@@ -543,11 +540,11 @@ SEXP rnng_messenger(SEXP url) {
     R_Free(sock);
     return mk_error(xc);
   }
-  dlp = R_Calloc(1, nng_listener);
+  dlp = (nng_listener *) R_Calloc(1, nng_listener);
   xc = nng_listen(*sock, up, dlp, 0);
   if (xc == 10 || xc == 15) {
     R_Free(dlp);
-    dlp = R_Calloc(1, nng_dialer);
+    dlp = (nng_dialer *) R_Calloc(1, nng_dialer);
     xc = nng_dial(*sock, up, dlp, 2u);
     if (xc) {
       R_Free(dlp);

tests/tests.R

@@ -87,7 +87,8 @@ nanotest(is.integer(ctx$id))
 nanotest(ctx$state == "opened")
 nanotest(ctx$protocol == "req")
 nanotest(setopt(ctx, "ms", "send-timeout", 1000) == 0L)
-nanotestwn(send_aio(ctx, data.frame(), timeout = 500))
+csaio <- send_aio(ctx, data.frame(), timeout = 500)
+nanotestn(call_aio(csaio)$result)
 nanotestwn(recv_aio(ctx1, timeout = 500, keep.raw = FALSE))
 nanotestwn(send(ctx, "context test", mode ="raw", block = 300, echo = FALSE))
 nanotestwn(recv(ctx1, mode = "character", block = 300))