From: Frederik Vanrenterghem Date: Fri, 14 Sep 2018 11:56:50 +0000 (+0800) Subject: Break out ConsumeKafkaMessages function. X-Git-Url: http://git.vanrenterghem.biz/R/project-using-kafka-in-R.git/commitdiff_plain/0ce8ac9c84fc8a3f79d46469a9c3c37232bb62f5 Break out ConsumeKafkaMessages function. --- 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)