summaryrefslogtreecommitdiffstats
path: root/python/fatcat_tools/workers
diff options
context:
space:
mode:
authorBryan Newbold <bnewbold@robocracy.org>2018-11-15 12:22:11 -0800
committerBryan Newbold <bnewbold@robocracy.org>2018-11-15 12:22:11 -0800
commit9f817c6c70a749f2ac449ab4edfd26c6dd8a7410 (patch)
tree10a430007ece7857d1799dec3d3324d654927abe /python/fatcat_tools/workers
parentf21d28315aa632cdb9f84ea8787762d1e27b4310 (diff)
downloadfatcat-9f817c6c70a749f2ac449ab4edfd26c6dd8a7410.tar.gz
fatcat-9f817c6c70a749f2ac449ab4edfd26c6dd8a7410.zip
have recent message helper cleanup consumer
Diffstat (limited to 'python/fatcat_tools/workers')
-rw-r--r--python/fatcat_tools/workers/worker_common.py6
1 files changed, 5 insertions, 1 deletions
diff --git a/python/fatcat_tools/workers/worker_common.py b/python/fatcat_tools/workers/worker_common.py
index daae91b3..e400e815 100644
--- a/python/fatcat_tools/workers/worker_common.py
+++ b/python/fatcat_tools/workers/worker_common.py
@@ -27,13 +27,17 @@ def most_recent_message(topic):
for p, op in consumer._partitions.items()]
offsets = [(p, (o if o > -1 else -2)) for p, o in offsets]
if -2 in [o for p, o in offsets]:
+ consumer.stop()
return None
else:
consumer.reset_offsets(offsets)
msg = islice(consumer, 1)
if msg:
- return list(msg)[0].value
+ val = list(msg)[0].value
+ consumer.stop()
+ return val
else:
+ consumer.stop()
return None
class FatcatWorker: