From 5b2ed10b2cd8f0326a38e9658a3d5cecf2849c6e Mon Sep 17 00:00:00 2001
From: Frederik Vanrenterghem <frederik@vanrenterghem.io>
Date: Fri, 14 Sep 2018 20:04:58 +0800
Subject: [PATCH] Break out SubscribeKafkaTopic function.

---
 R/SubscribeKafkaTopic.R | 12 ++++++++++++
 kafkaConsumer.R         | 10 ++--------
 2 files changed, 14 insertions(+), 8 deletions(-)
 create mode 100644 R/SubscribeKafkaTopic.R

diff --git a/R/SubscribeKafkaTopic.R b/R/SubscribeKafkaTopic.R
new file mode 100644
index 0000000..5e6976b
--- /dev/null
+++ b/R/SubscribeKafkaTopic.R
@@ -0,0 +1,12 @@
+SubscribeKafkaTopic <- function(consumer.base.uri, topic) {
+  response <- POST(url=paste(consumer.base.uri,
+                             "subscription", sep="/"),
+                   content_type("application/vnd.kafka.v2+json"),
+                   body = paste0('{"topics":["',
+                                 topic,
+                                 '"]}')
+  )
+  if(response$status_code == 204) {
+    return(response)
+  } else stop(response)
+}
\ No newline at end of file
diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index d6b092b..969fabd 100644
--- a/kafkaConsumer.R
+++ b/kafkaConsumer.R
@@ -19,14 +19,8 @@ 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,
-                           "subscription", sep="/"),
-               content_type("application/vnd.kafka.v2+json"),
-               body = paste0('{"topics":["',
-                             topic,
-                             '"]}')
-               )
-response
+source("R/SubscribeKafkaTopic.R")
+response <- SubscribeKafkaTopic(consumerDetails$base_uri,topic = topic)
 
 # Obtain all (or latest) messages on the topic
 source("R/ConsumeKafkaMessages.R")
-- 
2.39.5