Break out SubscribeKafkaTopic function.

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

diff --git a/kafkaConsumer.R b/kafkaConsumer.R
index a2ca952cc03dc55d75a77dfc600065973fb5e261..969fabd4f8dc42f34cd70adf27b87dfdcb608329 100644 (file)
--- a/kafkaConsumer.R
+++ b/kafkaConsumer.R
@@ -7,38 +7,31 @@ library(jsonlite)
 library(dplyr)
 
 kafka_rest_proxy <- "http://localhost:8082"
+consumer <- 'my_json_consumer'
+consumer_instance <- 'my_consumer_instance'
+topic <- 'accesslogapache'
 
 # Create consumer
 # See https://docs.confluent.io/current/kafka-rest/docs/intro.html#produce-and-consume-avro-messages
 
-response <- POST(url=paste(kafka_rest_proxy, "consumers", "my_json_consumer", sep="/"),
-     content_type("application/vnd.kafka.v2+json"),
-     accept("application/vnd.kafka.v2+json"),
-     body='{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}')
-fromJSON(content(response, "text"))
-
-response <- POST(url=paste(kafka_rest_proxy,
-                           "consumers",
-                           "my_json_consumer",
-                           "instances",
-                           "my_consumer_instance",
-                           "subscription", sep="/"),
-               content_type("application/vnd.kafka.v2+json"),
-               body = '{"topics":["accesslogapache"]}')
-response
-
-# Obtain all messages on the topic
-messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance","records", sep = "/"),
-                accept("application/vnd.kafka.json.v2+json"),
-                encode="json")
+# Create consumer
+source("R/CreateKafkaConsumer.R")
+consumerDetails <- CreateKafkaConsumer(kafka.rest.proxy = kafka_rest_proxy, consumer = consumer, consumer_instance = consumer_instance)
+
+# Subscribe it to topic
+source("R/SubscribeKafkaTopic.R")
+response <- SubscribeKafkaTopic(consumerDetails$base_uri,topic = topic)
 
-messages <- fromJSON(content(messagesJSON,"text"))
+# Obtain all (or latest) messages on the topic
+source("R/ConsumeKafkaMessages.R")
+messages <- ConsumeKafkaMessages(consumerDetails$base_uri)
 
 createPlot(messages$value)
 
 Sys.sleep(120)
 
-messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance","records", sep = "/"),
+# Obtain latest messages
+messagesJSON <- GET(url = paste(consumerDetails$base_uri,"records", sep = "/"),
                 accept("application/vnd.kafka.json.v2+json"),
                 encode="json")
 
@@ -49,5 +42,5 @@ apachelog <- rbind(messages$value,messages2$value)
 createPlot(apachelog)
 
 # Remove the consumer
-DELETE(url = paste(kafka_rest_proxy,"consumers","my_json_consumer","instances","my_consumer_instance", sep = "/"),
-       content_type("application/vnd.kafka.v2+json"))
+source("R/DestroyKafkaConsumer.R")
+response <- DestroyKafkaConsumer(consumer.base.uri = consumerDetails$base_uri)