diff options
| author | Mark Roberts <wizzat@gmail.com> | 2014-05-07 03:05:50 -0700 |
|---|---|---|
| committer | Mark Roberts <wizzat@gmail.com> | 2014-05-07 03:05:50 -0700 |
| commit | 39796ec49162a0895be98a8aeb0e42e8319e5e30 (patch) | |
| tree | 203e4c3dded544f2e70c1bbf142c5a0680163c87 /kafka/protocol.py | |
| parent | b47bf781eb0e96c3fef59cbe554325155062e129 (diff) | |
| parent | 671b74ab2e035a2f2ba7f90419794f6dbec08366 (diff) | |
| download | kafka-python-39796ec49162a0895be98a8aeb0e42e8319e5e30.tar.gz | |
Merge branch 'teach_producers_about_compression' into producer_compression
Conflicts:
servers/0.8.0/kafka-src
test/test_unit.py
Diffstat (limited to 'kafka/protocol.py')
| -rw-r--r-- | kafka/protocol.py | 23 |
1 files changed, 12 insertions, 11 deletions
diff --git a/kafka/protocol.py b/kafka/protocol.py index 7ec7946..730ae6b 100644 --- a/kafka/protocol.py +++ b/kafka/protocol.py @@ -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 # ################### @@ -150,17 +151,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) @@ -543,7 +544,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) @@ -564,6 +565,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) |