diff options
author | David Arthur <mumrah@gmail.com> | 2012-10-02 12:14:48 -0400 |
---|---|---|
committer | David Arthur <mumrah@gmail.com> | 2012-10-02 12:14:48 -0400 |
commit | c99d4119f2d6d0fdee38a159e1c845e42cf398b1 (patch) | |
tree | 3def57706d34be7c756ff94a37260e4a48382a52 /kafka | |
parent | 36b67b79202f8bdeab60de1f4ed582328eeb04db (diff) | |
download | kafka-python-c99d4119f2d6d0fdee38a159e1c845e42cf398b1.tar.gz |
Moved codec stuff into it's own module
Snappy will go there when I get around to it
Diffstat (limited to 'kafka')
-rw-r--r-- | kafka/__init__.py | 2 | ||||
-rw-r--r-- | kafka/codec.py | 23 | ||||
-rw-r--r-- | kafka/kafka.py | 23 |
3 files changed, 26 insertions, 22 deletions
diff --git a/kafka/__init__.py b/kafka/__init__.py index 69bc75f..1e01ea9 100644 --- a/kafka/__init__.py +++ b/kafka/__init__.py @@ -1 +1 @@ -__all__ = ["kafka"] +__all__ = ["kafka","codec"] diff --git a/kafka/codec.py b/kafka/codec.py new file mode 100644 index 0000000..47ab074 --- /dev/null +++ b/kafka/codec.py @@ -0,0 +1,23 @@ +from cStringIO import StringIO +import gzip +import logging + +log = logging.getLogger("kafka.codec") + +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 + +def gzip_decode(payload): + buf = StringIO(payload) + f = gzip.GzipFile(fileobj=buf, mode='r') + out = f.read() + f.close() + buf.close() + return out diff --git a/kafka/kafka.py b/kafka/kafka.py index ff9f53d..0cde87f 100644 --- a/kafka/kafka.py +++ b/kafka/kafka.py @@ -7,8 +7,9 @@ import socket import struct import zlib -log = logging.getLogger("org.apache.kafka") +from .codec import gzip_encode, gzip_decode +log = logging.getLogger("kafka") error_codes = { -1: "UnknownError", @@ -25,31 +26,11 @@ class KafkaException(Exception): def __str__(self): return str(errorType) - Message = namedtuple("Message", ["magic", "attributes", "crc", "payload"]) FetchRequest = namedtuple("FetchRequest", ["topic", "partition", "offset", "size"]) ProduceRequest = namedtuple("ProduceRequest", ["topic", "partition", "messages"]) OffsetRequest = namedtuple("OffsetRequest", ["topic", "partition", "time", "maxOffsets"]) -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 - -def gzip_decode(payload): - buf = StringIO(payload) - f = gzip.GzipFile(fileobj=buf, mode='r') - out = f.read() - f.close() - buf.close() - return out - - def length_prefix_message(msg): """ Prefix a message with it's length as an int |