Break out SubscribeKafkaTopic function.

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

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