From 7fa65fd2e51d4d96b6e5f895c29819c5d9ad31fb Mon Sep 17 00:00:00 2001
From: Frederik Vanrenterghem <frederik@vanrenterghem.io>
Date: Thu, 13 Sep 2018 21:13:42 +0800
Subject: [PATCH] Break out CreateKafkaConsumer function.

---
 R/CreateKafkaConsumer.R | 11 +++++++++++
 kafkaConsumer.R         | 10 ++--------
 2 files changed, 13 insertions(+), 8 deletions(-)
 create mode 100644 R/CreateKafkaConsumer.R

diff --git a/R/CreateKafkaConsumer.R b/R/CreateKafkaConsumer.R
new file mode 100644
index 0000000..cf6edee
--- /dev/null
+++ b/R/CreateKafkaConsumer.R
@@ -0,0 +1,11 @@
+CreateKafkaConsumer <- function(kafka.rest.proxy = "http://localhost:8082", consumer, consumer_instance) {
+  response <- POST(url=paste(kafka_rest_proxy, "consumers", consumer, sep="/"),
+                   content_type("application/vnd.kafka.v2+json"),
+                   accept("application/vnd.kafka.v2+json"),
+                   body=paste0('{"name": "',
+                               consumer_instance,
+                               '", "format": "json", "auto.offset.reset": "earliest"}')
+  )
+  consumerDetails <- fromJSON(content(response, "text"))
+  return(consumerDetails)
+}
\ No newline at end of file
diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index e7da6ed..6e8fe0b 100644
--- a/kafkaConsumer.R
+++ b/kafkaConsumer.R
@@ -15,14 +15,8 @@ topic <- 'accesslogapache'
 # See https://docs.confluent.io/current/kafka-rest/docs/intro.html#produce-and-consume-avro-messages
 
 # Create consumer
-response <- POST(url=paste(kafka_rest_proxy, "consumers", consumer, sep="/"),
-     content_type("application/vnd.kafka.v2+json"),
-     accept("application/vnd.kafka.v2+json"),
-     body=paste0('{"name": "',
-                 consumer_instance,
-                 '", "format": "json", "auto.offset.reset": "earliest"}')
-     )
-consumerDetails <- fromJSON(content(response, "text"))
+source("R/CreateKafkaConsumer.R")
+consumerDetails <- CreateKafkaConsumer(kafka.rest.proxy = kafka_rest_proxy, consumer = consumer, consumer_instance = consumer_instance)
 
 # Subscribe it to topic
 response <- POST(url=paste(consumerDetails$base_uri,
-- 
2.30.2