R/ConsumeKafkaMessages.R

   1 ConsumeKafkaMessages <- function(consumer.base.uri) {
   2   messagesJSON <- GET(url = paste(consumer.base.uri,"records", sep = "/"),
   3                       accept("application/vnd.kafka.json.v2+json"),
   4                       encode="json")
   5   if(messagesJSON$status_code!=200) stop(messagesJSON) else
   6   messages <- fromJSON(content(messagesJSON,"text"))
   7   return(messages)
   8 }