X-Git-Url: http://git.vanrenterghem.biz/R/project-using-kafka-in-R.git/blobdiff_plain/7fa65fd2e51d4d96b6e5f895c29819c5d9ad31fb..5b2ed10b2cd8f0326a38e9658a3d5cecf2849c6e:/kafkaConsumer.R

diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index 6e8fe0b..969fabd 100644
--- a/kafkaConsumer.R
+++ b/kafkaConsumer.R
@@ -19,21 +19,12 @@ 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
-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)
 
@@ -51,5 +42,5 @@ apachelog <- rbind(messages$value,messages2$value)
 createPlot(apachelog)
 
 # Remove the consumer
-DELETE(url = consumerDetails$base_uri,
-       content_type("application/vnd.kafka.v2+json"))
+source("R/DestroyKafkaConsumer.R")
+response <- DestroyKafkaConsumer(consumer.base.uri = consumerDetails$base_uri)