Merge branch 'issue-28'

Closes dpkp#28

Author: mumrah

.gitignore

@@ -1,2 +1,6 @@
+*.egg-info
 *.pyc
+.tox
 build
+dist
+MANIFEST

README.md

@@ -101,16 +101,12 @@ pip install python-snappy
 
 # Tests
 
-Some of the tests will fail if Snappy is not installed. These tests will throw
-NotImplementedError. If you see other failures, they might be bugs - so please
-report them!
-
 ## Run the unit tests
 
 _These are broken at the moment_
 
 ```shell
-python -m test.unit
+tox ./test/test_unit.py
 ```
 
 ## Run the integration tests
@@ -125,15 +121,9 @@ cd kafka-src
 ./sbt package
 ```
 
-Next start up a ZooKeeper server on localhost:2181
-
-```shell
-/opt/zookeeper/bin/zkServer.sh start
-```
-
-And finally run the tests. This will actually start up real Kafka brokers and send messages in using the
-client.
+And then run the tests. This will actually start up real local Zookeeper
+instance and Kafka brokers, and send messages in using the client.
 
 ```shell
-python -m test.integration
+tox ./test/test_integration.py
 ```

kafka/codec.py

@@ -1,44 +1,48 @@
 from cStringIO import StringIO
 import gzip
-import logging
-
-log = logging.getLogger("kafka.codec")
 
 try:
     import snappy
-    hasSnappy = True
+    _has_snappy = True
 except ImportError:
-    log.warn("Snappy codec not available")
-    hasSnappy = False
+    _has_snappy = False
+
+
+def has_gzip():
+    return True
+
+
+def has_snappy():
+    return _has_snappy
 
 
 def gzip_encode(payload):
-    buf = StringIO()
-    f = gzip.GzipFile(fileobj=buf, mode='w', compresslevel=6)
-    f.write(payload)
-    f.close()
-    buf.seek(0)
-    out = buf.read()
-    buf.close()
-    return out
+    buffer = StringIO()
+    handle = gzip.GzipFile(fileobj=buffer, mode="w")
+    handle.write(payload)
+    handle.close()
+    buffer.seek(0)
+    result = buffer.read()
+    buffer.close()
+    return result
 
 
 def gzip_decode(payload):
-    buf = StringIO(payload)
-    f = gzip.GzipFile(fileobj=buf, mode='r')
-    out = f.read()
-    f.close()
-    buf.close()
-    return out
+    buffer = StringIO(payload)
+    handle = gzip.GzipFile(fileobj=buffer, mode='r')
+    result = handle.read()
+    handle.close()
+    buffer.close()
+    return result
 
 
 def snappy_encode(payload):
-    if not hasSnappy:
-        raise NotImplementedError("Snappy codec not available")
+    if not _has_snappy:
+        raise NotImplementedError("Snappy codec is not available")
     return snappy.compress(payload)
 
 
 def snappy_decode(payload):
-    if not hasSnappy:
-        raise NotImplementedError("Snappy codec not available")
+    if not _has_snappy:
+        raise NotImplementedError("Snappy codec is not available")
     return snappy.decompress(payload)

kafka/consumer.py

@@ -83,6 +83,11 @@ def get_or_init_offset_callback(resp):
         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

setup.py

@@ -1,13 +1,41 @@
-from distutils.core import setup
+import os.path
+import sys
+
+from setuptools import setup, Command
+
+
+class Tox(Command):
+    user_options = []
+    def initialize_options(self):
+        pass
+
+    def finalize_options(self):
+        pass
+
+    def run(self):
+        import tox
+        sys.exit(tox.cmdline([]))
+
 
 setup(
     name="kafka-python",
     version="0.8.1-1",
+
+    install_requires=["distribute", "tox"],
+    tests_require=["tox"],
+    cmdclass={"test": Tox},
+
+    packages=["kafka"],
+
     author="David Arthur",
     author_email="[email protected]",
     url="https://github.com/mumrah/kafka-python",
-    packages=["kafka"],
     license="Copyright 2012, David Arthur under Apache License, v2.0",
     description="Pure Python client for Apache Kafka",
-    long_description=open("README.md").read(),
+    long_description="""
+This module provides low-level protocol support for Apache Kafka as well as
+high-level consumer and producer classes. Request batching is supported by the
+protocol as well as broker-aware request routing. Gzip and Snappy compression
+is also supported for message sets.
+"""
 )