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

# 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"}')
     )
fromJSON(content(response, "text"))

# Subscribe it to topic
response <- POST(url=paste(kafka_rest_proxy,
                           "consumers",
                           consumer,
                           "instances",
                           consumer_instance,
                           "subscription", sep="/"),
               content_type("application/vnd.kafka.v2+json"),
               body = paste0('{"topics":["',
                             topic,
                             '"]}')
               )
response

# Obtain all (or latest) messages on the topic
messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",consumer_instance,"records", sep = "/"),
                accept("application/vnd.kafka.json.v2+json"),
                encode="json")

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

createPlot(messages$value)

Sys.sleep(120)

# Obtain latest messages
messagesJSON <- GET(url = paste(kafka_rest_proxy,"consumers",consumer,"instances",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",consumer,"instances",consumer_instance, sep = "/"),
       content_type("application/vnd.kafka.v2+json"))