Add support and instructions for development against locally running queues

DESCRIPTION

@@ -11,13 +11,16 @@ License: MIT + file LICENSE
 LazyData: TRUE
 Depends: 
   R (>= 3.0.0),
-  pkgdown,
-  aws.sqs
+  pkgdown
 Imports:
-  jsonlite
+  jsonlite,
+  aws.signature,
+  jsonlite,
+  purrr,
+  xml2
 Remotes:
   datacamp/pkgdown
 Suggests:
   testthat
-RoxygenNote: 6.0.1
+RoxygenNote: 7.1.1
 VignetteBuilder: knitr

Dockerfile

@@ -15,7 +15,6 @@ RUN apt-get install libharfbuzz-dev libfribidi-dev
 
 RUN R -e 'install.packages("remotes")'
 RUN R -e 'remotes::install_github("datacamp/pkgdown", ref = "master")'
-RUN R -e 'install.packages("aws.sqs", repos = c(getOption("repos"), "http://cloudyr.github.io/drat"))'
 
 COPY . r-package-parser
 

NAMESPACE

@@ -1,16 +1,22 @@
 # Generated by roxygen2: do not edit by hand
 
+export(consume_msg)
+export(delete_msg)
 export(main)
 export(parse_description)
 export(parse_topics)
 export(process_package)
-importFrom(aws.sqs,create_queue)
-importFrom(aws.sqs,delete_msg)
-importFrom(aws.sqs,receive_msg)
+export(receive_msg)
+export(send_msg)
+export(sqsHTTP)
+import(httr)
+importFrom(aws.signature,signature_v4_auth)
 importFrom(jsonlite,fromJSON)
 importFrom(jsonlite,prettify)
 importFrom(jsonlite,toJSON)
 importFrom(jsonlite,write_json)
 importFrom(magrittr,"%>%")
 importFrom(purrr,map)
 importFrom(purrr,transpose)
+importFrom(xml2,as_list)
+importFrom(xml2,read_xml)

R/aws.R

@@ -37,53 +37,13 @@ copy_local <- function(local, path, dirname){
   }
 }
 
-send_msg <- function(queue, msg, query = list(), attributes = NULL, delay = NULL, ...) {
-  queue <- aws.sqs:::.urlFromName(queue)
-  if(length(msg) > 1) {
-    # batch mode
-    batchs <- split(msg, ceiling(seq_along(msg)/10))
+send_msg_wrap <- function(queue, msg, type) {
+  info(paste(sprintf("Sending %s message to %s", type, queue), prettify(msg), sep="\n"))
 
-    for (batch in batchs) {
-      l <- length(batch)
-      n <- 1:l
-      id <- paste0("msg", n)
-      a <- as.list(c(id, batch))
-      names(a) <- c(paste0("SendMessageBatchRequestEntry.",n,".Id"),
-                    paste0("SendMessageBatchRequestEntry.",n,".MessageBody"))
-      query_args <-  list(Action = "SendMessageBatch")
-      query_mult <- rep(query, each = l)
-      front <- c(paste0("SendMessageBatchRequestEntry.",n, "."))
-      back <- rep(names(query), each = l)
-      names(query_mult) <- paste0(front, back)
-
-      body <- c(a, query_mult, query_args)
-
-      out <- aws.sqs:::sqsHTTP(url = queue, query = body, ...)
-      if (inherits(out, "aws-error") || inherits(out, "unknown")) {
-        return(out)
-      }
-      structure(out$SendMessageBatchResponse$SendMessageBatchResult,
-                RequestId = out$SendMessageBatchResponse$ResponseMetadata$RequestId)
-    }
-
-  } else {
-    # single mode
-    query_args <- append(query, list(Action = "SendMessage"))
-    query_args$MessageBody = msg
-    out <- aws.sqs:::sqsHTTP(url = queue, query = query_args, ...)
-    if (inherits(out, "aws-error") || inherits(out, "unknown")) {
-      return(out)
-    }
-    structure(list(out$SendMessageResponse$SendMessageResult),
-              RequestId = out$SendMessageResponse$ResponseMetadata$RequestId)
-  }
-}
-
-post_job <- function(queue, json, value) {
-  info(sprintf("Posting %s job...", value))
-  send_msg(queue,
-           msg = json,
+  send_msg(queue = queue,
+           msg = msg,
            query = list(MessageAttribute.1.Name = "type",
                         MessageAttribute.1.Value.DataType ="String",
-                        MessageAttribute.1.Value.StringValue = value))
+                        MessageAttribute.1.Value.StringValue = type))
 }
+

R/aws_sqs_attributes.R

@@ -0,0 +1,37 @@
+#' @title Get a queue URL
+#' @aliases get_queue_url
+#' @description Retrieves the URL for an SQS queue by its name.
+#' @param name A character string containing the name of the queue.
+#' @param owner A character string containing the AWS Account ID that created the queue.
+#' @param query A list specifying additional query arguments to be passed to the \code{query} argument of \code{\link{sqsHTTP}}.
+#' @param ... Additional arguments passed to \code{\link{sqsHTTP}}.
+#' @return If successful, a character string containing an SQS Queue URL. Otherwise, a data structure of class \dQuote{aws_error} containing any error message(s) from AWS and information about the request attempt.
+#' @author Thomas J. Leeper
+#' @seealso \code{link{create_queue}} \code{link{delete_queue}} \code{\link{get_queue_attrs}} \code{\link{set_queue_attrs}}
+#' @references
+#' \href{http://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_GetQueueUrl.html}{GetQueueURL}
+#' @export
+get_queue_url <- function(name, owner = NULL, query = NULL, ...) {
+  query_args <- c(query, list(Action = "GetQueueUrl", QueueName = name))
+  if (!is.null(owner)) {
+    query_args$QueueOwnerAWSAccountId <- owner
+  }
+  out <- sqsHTTP(query = query_args, ...)
+  if (inherits(out, "aws-error") || inherits(out, "unknown")) {
+    return(out)
+  }
+  structure(out$GetQueueUrlResponse$GetQueueUrlResult$QueueUrl,
+            RequestId = out$GetQueueUrlResponse$ResponseMetadata$RequestId)
+}
+
+.urlFromName <- function(queue) {
+  p <- httr::parse_url(queue)
+  if (is.null(p$scheme)) {
+    out <- get_queue_url(queue)
+    if(!length(out))
+      stop("Queue URL not found")
+  } else {
+    out <- queue
+  }
+  return(out)
+}

R/aws_sqs_http.R

@@ -0,0 +1,102 @@
+#' @title Execute SQS API Request
+#' @description This is the workhorse function to execute calls to the SQS API.
+#' @details This function constructs and signs an SQS API request and returns the results thereof, or relevant debugging information in the case of error.
+#' @param url A character string containing an SQS API endpoint URL.
+#' @param query An optional named list containing query string parameters and their character values.
+#' @param headers A list of headers to pass to the HTTP request.
+#' @param verbose A logical indicating whether to be verbose. Default is given by \code{options("verbose")}.
+#' @param region A character string specifying an AWS region. See \code{\link[aws.signature]{locate_credentials}}.
+#' @param key A character string specifying an AWS Access Key. See \code{\link[aws.signature]{locate_credentials}}.
+#' @param secret A character string specifying an AWS Secret Key. See \code{\link[aws.signature]{locate_credentials}}.
+#' @param session_token Optionally, a character string specifying an AWS temporary Session Token to use in signing a request. See \code{\link[aws.signature]{locate_credentials}}.
+#' @param ... Additional arguments passed to \code{\link[httr]{GET}}.
+#' @return If successful, a named list. Otherwise, a data structure of class
+#' \dQuote{aws_error} containing any error message(s) from AWS and information
+#' about the request attempt.
+#' @author Thomas J. Leeper
+#' @importFrom aws.signature signature_v4_auth
+#' @importFrom jsonlite fromJSON
+#' @importFrom xml2 read_xml as_list
+#' @import httr
+#' @export
+sqsHTTP <-
+  function(
+    url = NULL,
+    headers = list(),
+    query = list(),
+    verbose = getOption("verbose", FALSE),
+    region = Sys.getenv("AWS_DEFAULT_REGION", "us-east-1"),
+    key = NULL,
+    secret = NULL,
+    session_token = NULL,
+    ...
+  ) {
+    # locate and validate credentials
+    credentials <- aws.signature::locate_credentials(key = key, secret = secret, session_token = session_token, region = region, verbose = verbose)
+    key <- credentials[["key"]]
+    secret <- credentials[["secret"]]
+    session_token <- credentials[["session_token"]]
+    region <- credentials[["region"]]
+
+    # generate request signature
+    if (is.null(url)) {
+      url <- paste0("https://sqs.",region,".amazonaws.com")
+    }
+    p <- httr::parse_url(url)
+    action <- if(p$path == "") "/" else paste0("/", p$path)
+    d_timestamp <- format(Sys.time(), "%Y%m%dT%H%M%SZ", tz = "UTC")
+    Sig <- aws.signature::signature_v4_auth(
+      datetime = d_timestamp,
+      region = region,
+      service = "sqs",
+      verb = "GET",
+      action = action,
+      query_args = query,
+      canonical_headers = list(host = paste0("sqs.",region,".amazonaws.com"),
+                               `x-amz-date` = d_timestamp),
+      request_body = "",
+      key = key,
+      secret = secret,
+      session_token = session_token,
+      verbose = verbose)
+    # setup request headers
+    headers[["x-amz-date"]] <- d_timestamp
+    headers[["x-amz-content-sha256"]] <- Sig$BodyHash
+    headers[["Authorization"]] <- Sig[["SignatureHeader"]]
+    if (!is.null(session_token) && session_token != "") {
+      headers[["x-amz-security-token"]] <- session_token
+    }
+    H <- do.call(httr::add_headers, headers)
+
+    # execute request
+    if (length(query)) {
+      r <- httr::GET(url, H, query = query, ...)
+    } else {
+      r <- httr::GET(url, H, ...)
+    }
+
+    cont <- httr::content(r, "text", encoding = "UTF-8")
+    if (httr::http_error(r)) {
+      x <- try(xml2::as_list(xml2::read_xml(cont)), silent = TRUE)
+      if (inherits(x, "try-error")) {
+        x <- try(jsonlite::fromJSON(cont)$Error, silent = TRUE)
+      }
+      warning(paste0(httr::http_status(r)$message, ": ", x$Code, " (", x$Message, ")"))
+      h <- httr::headers(r)
+      out <- structure(x, headers = h, class = "aws_error")
+      attr(out, "request_canonical") <- Sig$CanonicalRequest
+      attr(out, "request_string_to_sign") <- Sig$StringToSign
+      attr(out, "request_signature") <- Sig$SignatureHeader
+    } else {
+      out <- try(jsonlite::fromJSON(cont, simplifyVector = FALSE), silent = TRUE)
+      if (inherits(out, "try-error")) {
+        out2 <- try(xml2::as_list(xml2::read_xml(cont)), silent = TRUE)
+        if (inherits(out2, "try-error")) {
+          out <- structure(cont, class = "unknown")
+        } else {
+          out <- out2
+        }
+      }
+    }
+    return(out)
+  }