From: Frederik Vanrenterghem Date: Thu, 13 Sep 2018 13:13:42 +0000 (+0800) Subject: Break out CreateKafkaConsumer function. X-Git-Url: http://git.vanrenterghem.biz/R/project-using-kafka-in-R.git/commitdiff_plain/7fa65fd2e51d4d96b6e5f895c29819c5d9ad31fb?hp=ad70a446e1f9c1d371008dc32f7374a7e3a54907 Break out CreateKafkaConsumer function. --- diff --git a/R/CreateKafkaConsumer.R b/R/CreateKafkaConsumer.R new file mode 100644 index 0000000..cf6edee --- /dev/null +++ b/R/CreateKafkaConsumer.R @@ -0,0 +1,11 @@ +CreateKafkaConsumer <- function(kafka.rest.proxy = "http://localhost:8082", consumer, consumer_instance) { + 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")) + return(consumerDetails) +} \ No newline at end of file diff --git a/kafkaConsumer.R b/kafkaConsumer.R index e7da6ed..6e8fe0b 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,