Break out ConsumeKafkaMessages function.

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

diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index 8b5e6c7f8acee1edb8c53debed53cc37fb002030..d6b092b917a6af028e2725f4876790ef317b203b 100644 (file)
--- 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)