Coverage for portality/events/kafka_producer.py: 0%

1import json 

2from kafka import KafkaProducer 

3 

4from portality.core import app as doajapp 

5bootstrap_server = doajapp.config.get("KAFKA_BOOTSTRAP_SERVER") 

6 

7producer = KafkaProducer(bootstrap_servers=bootstrap_server, value_serializer=lambda v: json.dumps(v).encode('utf-8')) 

8 

9 

10def send_event(event): 

11 future = producer.send('events', value=event.serialise()) 

12 future.get(timeout=60)