diff options
| author | Bryan Newbold <bnewbold@robocracy.org> | 2018-11-15 12:22:11 -0800 |
|---|---|---|
| committer | Bryan Newbold <bnewbold@robocracy.org> | 2018-11-15 12:22:11 -0800 |
| commit | 9f817c6c70a749f2ac449ab4edfd26c6dd8a7410 (patch) | |
| tree | 10a430007ece7857d1799dec3d3324d654927abe /python/fatcat_tools/workers | |
| parent | f21d28315aa632cdb9f84ea8787762d1e27b4310 (diff) | |
| download | fatcat-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.py | 6 |
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: |