diff options
Diffstat (limited to 'kafka')
| -rw-r--r-- | kafka/codec.py | 7 | ||||
| -rw-r--r-- | kafka/consumer.py | 5 |
2 files changed, 12 insertions, 0 deletions
diff --git a/kafka/codec.py b/kafka/codec.py index c7d3992..cb3858b 100644 --- a/kafka/codec.py +++ b/kafka/codec.py @@ -11,6 +11,13 @@ except ImportError: log.warn("Snappy codec not available") hasSnappy = False +def has_gzip(): + return True + + +def has_snappy(): + return hasSnappy + def gzip_encode(payload): buf = StringIO() diff --git a/kafka/consumer.py b/kafka/consumer.py index 3b64571..c9f12e1 100644 --- a/kafka/consumer.py +++ b/kafka/consumer.py @@ -83,6 +83,11 @@ class SimpleConsumer(object): for partition in self.client.topic_partitions[topic]: self.offsets[partition] = 0 + def stop(self): + if self.commit_timer is not None: + self.commit_timer.stop() + self.commit() + def seek(self, offset, whence): """ Alter the current offset in the consumer, similar to fseek |