amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-08-28 09:53:22 +00:00
Code Issues Projects Releases Wiki Activity

Modified to send entire ordered dict to Kafka. Bug: would barf on reports larger than 10 megs

Browse Source
This commit is contained in:
Mike Siegel
2018-10-10 08:07:44 -04:00
parent d4cf4a7e5f
commit a3ba85803a
1 changed files with 7 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parsedmarc/kafkaclient.py
+7 -5
View File
@@ -11,7 +11,8 @@ class KafkaClient(object):
def __init__(self, kafka_hosts):
""" Right now lets just do one host"""
self.host = kafka_hosts
self.producer = KafkaProducer(bootstrap_servers=kafka_hosts, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
self.producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'),
bootstrap_servers=kafka_hosts)
def save_aggregate_reports_to_kafka(self, aggregate_reports, aggregate_topic):
"""
@@ -28,9 +29,10 @@ class KafkaClient(object):
if len(aggregate_reports) < 1:
return
for report in aggregate_reports:
self.producer.send(aggregate_topic, report)
self.producer.flush()
print("Report is {}".format(aggregate_reports))
print("Report type is {}".format(type(aggregate_reports)))
self.producer.send(aggregate_topic, aggregate_reports)
self.producer.flush()
def save_forensic_reports_to_kafka(self, forensic_reports, forensic_topic):
@@ -49,5 +51,5 @@ class KafkaClient(object):
return
for report in forensic_reports:
self.producer.send(forensic_topic, report)
self.producer.send(forensic_topic, json.dumps(report))
self.producer.flush()