X-Git-Url: http://git.vanrenterghem.biz/R/project-using-kafka-in-R.git/blobdiff_plain/ad70a446e1f9c1d371008dc32f7374a7e3a54907..0ce8ac9c84fc8a3f79d46469a9c3c37232bb62f5:/kafkaConsumer.R

diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index e7da6ed..d6b092b 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,
@@ -35,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)
 
@@ -57,5 +48,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)