Break out ConsumeKafkaMessages function.

[R/project-using-kafka-in-R.git] / R / ConsumeKafkaMessages.R

diff --git a/R/ConsumeKafkaMessages.R b/R/ConsumeKafkaMessages.R
new file mode 100644 (file)
index 0000000..4013b5c
--- /dev/null
+++ b/R/ConsumeKafkaMessages.R
@@ -0,0 +1,8 @@
+ConsumeKafkaMessages <- function(consumer.base.uri) {
+  messagesJSON <- GET(url = paste(consumer.base.uri,"records", sep = "/"),
+                      accept("application/vnd.kafka.json.v2+json"),
+                      encode="json")
+  if(messagesJSON$status_code!=200) stop(messagesJSON) else
+  messages <- fromJSON(content(messagesJSON,"text"))
+  return(messages)
+}
\ No newline at end of file