kafka_cmd_wrappers.R

create_kafka_producer = function(topic, broker_list = "localhost:9092", open = "w") {
  prodocer_cmd = "kafka-console-producer"
  # kafka-console-producer --topic test2 --broker-list 
  kafka_pipe = sprintf("%s --topic %s --broker-list %s", prodocer_cmd, topic, broker_list)
  pipe(kafka_pipe, open = open)
}
# con = create_kafka_producer("test2", open = "wb")
# writeLines("5", con)
# flush(con)
# close.connection(con)

create_kafka_consumer = function(topic, bootstrap_server = "localhost:9092") {
  consumer_cmd = "kafka-console-consumer"
  kafka_pipe = sprintf("%s --topic %s --bootstrap-server %s --from-beginning", consumer_cmd, topic, bootstrap_server)
  pipe(kafka_pipe, open = "r")
}

# consumer_con = create_kafka_consumer("test2")
# messages = readLines(consumer_con)
# close.connection(consumer_con)