# Path: kiel/constants.py
# GZIP = 1
# 
# SNAPPY = 2
# 
# Path: kiel/compression/gzip.py
# def compress(data):
# def decompress(data):
# 
# Path: kiel/compression/snappy.py
# DEFAULT_VERSION = 1
# MIN_COMPAT_VERSION = 1
# MAGIC_HEADER = (
#     -126, b'S', b'N', b'A', b'P', b'P', b'Y', 0,
#     DEFAULT_VERSION, MIN_COMPAT_VERSION
# )
# BLOCK_SIZE = 32 * 1024  # 32kb, in bytes
# def compress(data):
# def decompress(data):
# 
# Path: kiel/protocol/part.py
# class Part(object):
#     """
#     Composable building block used to define Kafka protocol parts.
# 
#     Behaves much like the `Primitive` class but has named "sub parts"
#     stored in a ``parts`` class attribute, that can hold any `Part` or
#     `Primitive` subclass.
#     """
#     parts = ()
# 
#     def __init__(self, **kwargs):
#         part_names = [item[0] for item in self.parts]
# 
#         for name, value in kwargs.items():
#             if name not in part_names:
#                 raise ValueError("Unknown part name: '%s'" % name)
# 
#             setattr(self, name, value)
# 
#     def render(self, parts=None):
#         """
#         Returns a two-element tuple with the ``struct`` format and values.
# 
#         Iterates over the applicable sub-parts and calls `render()` on them,
#         accumulating the format string and values.
# 
#         Optionally takes a subset of parts to render, default behavior is to
#         render all sub-parts belonging to the class.
#         """
#         if not parts:
#             parts = self.parts
# 
#         fmt = []
#         data = []
# 
#         for name, part_class in parts:
#             if issubclass(part_class, Primitive):
#                 part = part_class(getattr(self, name, None))
#             else:
#                 part = getattr(self, name, None)
# 
#             part_format, part_data = part.render()
# 
#             fmt.extend(part_format)
#             data.extend(part_data)
# 
#         return "".join(fmt), data
# 
#     @classmethod
#     def parse(cls, buff, offset):
#         """
#         Given a buffer and offset, returns the parsed value and new offset.
# 
#         Calls `parse()` on the given buffer for each sub-part in order and
#         creates a new instance with the results.
#         """
#         values = {}
# 
#         for name, part in cls.parts:
#             value, new_offset = part.parse(buff, offset)
# 
#             values[name] = value
#             offset = new_offset
# 
#         return cls(**values), offset
# 
#     def __eq__(self, other):
#         """
#         `Part` instances are equal if all of their sub-parts are also equal.
#         """
#         try:
#             return all([
#                 getattr(self, part_name) == getattr(other, part_name)
#                 for part_name, part_class in self.parts
#             ])
#         except AttributeError:
#             return False
# 
#     def __repr__(self):
# 
#         def subpart_string(part_info):
#             part_name, part_class = part_info
# 
#             if not part_class.__name__.startswith("ArrayOf"):
#                 return "%s=%s" % (part_name, getattr(self, part_name, None))
# 
#             return "%s=[%s]" % (
#                 part_name,
#                 ", ".join([
#                     str(item) for item in getattr(self, part_name, [])
#                 ])
#             )
# 
#         return "%s(%s)" % (
#             self.__class__.__name__,
#             ", ".join([subpart_string(part) for part in self.parts])
#         )
# 
# Path: kiel/protocol/primitives.py
# class Int8(Primitive):
#     """
#     Represents an 8-bit signed integer.
#     """
#     fmt = "b"
# 
# class Int32(Primitive):
#     """
#     Represents an 32-bit signed integer.
#     """
#     fmt = "i"
# 
# class Int64(Primitive):
#     """
#     Represents an 64-bit signed integer.
#     """
#     fmt = "q"
# 
# class Bytes(VariablePrimitive):
#     """
#     Represents a bytestring value, length denoted by a 32-bit signed integer.
#     """
#     size_primitive = Int32


import logging
import struct
import zlib
from kiel.constants import GZIP, SNAPPY
from kiel.compression import gzip, snappy
from .part import Part
from .primitives import Int8, Int32, Int64, Bytes
        crc = zlib.crc32(payload)
        if crc > (2**31):
            crc -= 2**32

        fmt = "i%ds" % len(payload)

        return fmt, [crc, payload]

    @classmethod
    def parse(cls, buff, offset):
        """
        Given a buffer and offset, returns the parsed `Message` and new offset.

        Parses via the basic ``Part`` parse method, but with added
        decompression support.

        If a parsed message's attributes denote that compression has been used,
        the value is run through the corresponding ``decompress()`` method.
        """
        message, offset = super(Message, cls).parse(buff, offset)

        # the compression scheme is stored in the lowest 2 bits of 'attributes'
        compression = message.attributes & 0b00000011

        if not compression:
            return message, offset

        if compression == GZIP:
            message.value = gzip.decompress(message.value)
        elif compression == SNAPPY:

# Path: kiel/constants.py
# GZIP = 1
# 
# SNAPPY = 2
# 
# Path: kiel/compression/gzip.py
# def compress(data):
# def decompress(data):
# 
# Path: kiel/compression/snappy.py
# DEFAULT_VERSION = 1
# MIN_COMPAT_VERSION = 1
# MAGIC_HEADER = (
#     -126, b'S', b'N', b'A', b'P', b'P', b'Y', 0,
#     DEFAULT_VERSION, MIN_COMPAT_VERSION
# )
# BLOCK_SIZE = 32 * 1024  # 32kb, in bytes
# def compress(data):
# def decompress(data):
# 
# Path: kiel/protocol/part.py
# class Part(object):
#     """
#     Composable building block used to define Kafka protocol parts.
# 
#     Behaves much like the `Primitive` class but has named "sub parts"
#     stored in a ``parts`` class attribute, that can hold any `Part` or
#     `Primitive` subclass.
#     """
#     parts = ()
# 
#     def __init__(self, **kwargs):
#         part_names = [item[0] for item in self.parts]
# 
#         for name, value in kwargs.items():
#             if name not in part_names:
#                 raise ValueError("Unknown part name: '%s'" % name)
# 
#             setattr(self, name, value)
# 
#     def render(self, parts=None):
#         """
#         Returns a two-element tuple with the ``struct`` format and values.
# 
#         Iterates over the applicable sub-parts and calls `render()` on them,
#         accumulating the format string and values.
# 
#         Optionally takes a subset of parts to render, default behavior is to
#         render all sub-parts belonging to the class.
#         """
#         if not parts:
#             parts = self.parts
# 
#         fmt = []
#         data = []
# 
#         for name, part_class in parts:
#             if issubclass(part_class, Primitive):
#                 part = part_class(getattr(self, name, None))
#             else:
#                 part = getattr(self, name, None)
# 
#             part_format, part_data = part.render()
# 
#             fmt.extend(part_format)
#             data.extend(part_data)
# 
#         return "".join(fmt), data
# 
#     @classmethod
#     def parse(cls, buff, offset):
#         """
#         Given a buffer and offset, returns the parsed value and new offset.
# 
#         Calls `parse()` on the given buffer for each sub-part in order and
#         creates a new instance with the results.
#         """
#         values = {}
# 
#         for name, part in cls.parts:
#             value, new_offset = part.parse(buff, offset)
# 
#             values[name] = value
#             offset = new_offset
# 
#         return cls(**values), offset
# 
#     def __eq__(self, other):
#         """
#         `Part` instances are equal if all of their sub-parts are also equal.
#         """
#         try:
#             return all([
#                 getattr(self, part_name) == getattr(other, part_name)
#                 for part_name, part_class in self.parts
#             ])
#         except AttributeError:
#             return False
# 
#     def __repr__(self):
# 
#         def subpart_string(part_info):
#             part_name, part_class = part_info
# 
#             if not part_class.__name__.startswith("ArrayOf"):
#                 return "%s=%s" % (part_name, getattr(self, part_name, None))
# 
#             return "%s=[%s]" % (
#                 part_name,
#                 ", ".join([
#                     str(item) for item in getattr(self, part_name, [])
#                 ])
#             )
# 
#         return "%s(%s)" % (
#             self.__class__.__name__,
#             ", ".join([subpart_string(part) for part in self.parts])
#         )
# 
# Path: kiel/protocol/primitives.py
# class Int8(Primitive):
#     """
#     Represents an 8-bit signed integer.
#     """
#     fmt = "b"
# 
# class Int32(Primitive):
#     """
#     Represents an 32-bit signed integer.
#     """
#     fmt = "i"
# 
# class Int64(Primitive):
#     """
#     Represents an 64-bit signed integer.
#     """
#     fmt = "q"
# 
# class Bytes(VariablePrimitive):
#     """
#     Represents a bytestring value, length denoted by a 32-bit signed integer.
#     """
#     size_primitive = Int32


# Path: kiel/protocol/messages.py
import logging
import struct
import zlib
from kiel.constants import GZIP, SNAPPY
from kiel.compression import gzip, snappy
from .part import Part
from .primitives import Int8, Int32, Int64, Bytes

        crc = zlib.crc32(payload)
        if crc > (2**31):
            crc -= 2**32

        fmt = "i%ds" % len(payload)

        return fmt, [crc, payload]

    @classmethod
    def parse(cls, buff, offset):
        """
        Given a buffer and offset, returns the parsed `Message` and new offset.

        Parses via the basic ``Part`` parse method, but with added
        decompression support.

        If a parsed message's attributes denote that compression has been used,
        the value is run through the corresponding ``decompress()`` method.
        """
        message, offset = super(Message, cls).parse(buff, offset)

        # the compression scheme is stored in the lowest 2 bits of 'attributes'
        compression = message.attributes & 0b00000011

        if not compression:
            return message, offset

        if compression == GZIP:
            message.value = gzip.decompress(message.value)
        elif compression == SNAPPY:

            message.value = snappy.decompress(message.value)