openstack/deb-python-kafka
Code Issues Proposed changes

Add 'codec' parameter to Producer

Browse Source 
Adds a codec parameter to Producer.__init__ that lets the user choose
a compression codec to use for all messages sent by it.
This commit is contained in:
Patrick Lucas
2014-05-03 11:27:57 -07:00
parent 2415609ce0
commit 671b74ab2e
3 changed files with 49 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
kafka/protocol.py
View File

@@ -18,6 +18,12 @@ from kafka.util import (
log = logging.getLogger("kafka")
ATTRIBUTE_CODEC_MASK = 0x03
CODEC_NONE = 0x00
CODEC_GZIP = 0x01
CODEC_SNAPPY = 0x02
ALL_CODECS = (CODEC_NONE, CODEC_GZIP, CODEC_SNAPPY)
class KafkaProtocol(object):
"""
@@ -32,11 +38,6 @@ class KafkaProtocol(object):
OFFSET_COMMIT_KEY = 8
OFFSET_FETCH_KEY = 9
ATTRIBUTE_CODEC_MASK = 0x03
CODEC_NONE = 0x00
CODEC_GZIP = 0x01
CODEC_SNAPPY = 0x02
###################
# Private API #
###################
@@ -151,17 +152,17 @@ class KafkaProtocol(object):
(key, cur) = read_int_string(data, cur)
(value, cur) = read_int_string(data, cur)
codec = att & KafkaProtocol.ATTRIBUTE_CODEC_MASK
codec = att & ATTRIBUTE_CODEC_MASK
if codec == KafkaProtocol.CODEC_NONE:
if codec == CODEC_NONE:
yield (offset, Message(magic, att, key, value))
elif codec == KafkaProtocol.CODEC_GZIP:
elif codec == CODEC_GZIP:
gz = gzip_decode(value)
for (offset, msg) in KafkaProtocol._decode_message_set_iter(gz):
yield (offset, msg)
elif codec == KafkaProtocol.CODEC_SNAPPY:
elif codec == CODEC_SNAPPY:
snp = snappy_decode(value)
for (offset, msg) in KafkaProtocol._decode_message_set_iter(snp):
yield (offset, msg)
@@ -544,7 +545,7 @@ def create_gzip_message(payloads, key=None):
[create_message(payload) for payload in payloads])
gzipped = gzip_encode(message_set)
codec = KafkaProtocol.ATTRIBUTE_CODEC_MASK & KafkaProtocol.CODEC_GZIP
codec = ATTRIBUTE_CODEC_MASK & CODEC_GZIP
return Message(0, 0x00 | codec, key, gzipped)
@@ -565,6 +566,6 @@ def create_snappy_message(payloads, key=None):
[create_message(payload) for payload in payloads])
snapped = snappy_encode(message_set)
codec = KafkaProtocol.ATTRIBUTE_CODEC_MASK & KafkaProtocol.CODEC_SNAPPY
codec = ATTRIBUTE_CODEC_MASK & CODEC_SNAPPY
return Message(0, 0x00 | codec, key, snapped)