listen_kafka.py

"""Print Kafka events sent by the webservice.
"""
import json
import os.path as osp

import yaml
from kafka import KafkaConsumer

conf_file = osp.join(osp.dirname(__file__), 'webservice.yaml')

with open(conf_file, "r") as f:
    config = yaml.safe_load(f)

topic = config['kafka']['topic']
brokers = config['kafka']['brokers']
print(f"topic: {topic!r}")
print(f"brokers: {brokers!r}")

kc = KafkaConsumer(topic, bootstrap_servers=brokers)

print("Waiting for messages (Ctrl-C to quit)...")
try:
    for record in kc:
        msg = json.loads(record.value.decode())
        print(msg)
except KeyboardInterrupt:
    print("Exiting")
    kc.close()