From 0ce8ac9c84fc8a3f79d46469a9c3c37232bb62f5 Mon Sep 17 00:00:00 2001
From: Frederik Vanrenterghem <frederik@vanrenterghem.io>
Date: Fri, 14 Sep 2018 19:56:50 +0800
Subject: [PATCH] Break out ConsumeKafkaMessages function.

---
 R/ConsumeKafkaMessages.R | 8 ++++++++
 kafkaConsumer.R          | 7 ++-----
 2 files changed, 10 insertions(+), 5 deletions(-)
 create mode 100644 R/ConsumeKafkaMessages.R

diff --git a/R/ConsumeKafkaMessages.R b/R/ConsumeKafkaMessages.R
new file mode 100644
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
diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index 8b5e6c7..d6b092b 100644
--- a/kafkaConsumer.R
+++ b/kafkaConsumer.R
@@ -29,11 +29,8 @@ response <- POST(url=paste(consumerDetails$base_uri,
 response
 
 # Obtain all (or latest) messages on the topic
-messagesJSON <- GET(url = paste(consumerDetails$base_uri,"records", sep = "/"),
-                accept("application/vnd.kafka.json.v2+json"),
-                encode="json")
-
-messages <- fromJSON(content(messagesJSON,"text"))
+source("R/ConsumeKafkaMessages.R")
+messages <- ConsumeKafkaMessages(consumerDetails$base_uri)
 
 createPlot(messages$value)
 
-- 
2.39.5