summaryrefslogtreecommitdiff
path: root/kafka/kafka.py
diff options
context:
space:
mode:
authorDavid Arthur <mumrah@gmail.com>2012-10-02 12:14:48 -0400
committerDavid Arthur <mumrah@gmail.com>2012-10-02 12:14:48 -0400
commitc99d4119f2d6d0fdee38a159e1c845e42cf398b1 (patch)
tree3def57706d34be7c756ff94a37260e4a48382a52 /kafka/kafka.py
parent36b67b79202f8bdeab60de1f4ed582328eeb04db (diff)
downloadkafka-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/kafka.py')
-rw-r--r--kafka/kafka.py23
1 files changed, 2 insertions, 21 deletions
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