# Read via REST API from Kafka topic
# Prerequisite: set up kafka with topic accesslogapache


library(httr)
library(jsonlite)
library(dplyr)

kafka_rest_proxy <- "http://localhost:8082"

# 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")

messages <- fromJSON(content(messagesJSON,"text"))

createPlot(messages$value)

Sys.sleep(120)

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")

messages2 <- fromJSON(content(messagesJSON,"text"))

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"))