# Path: kiel/exc.py
# class BrokerConnectionError(KielError):
#     """
#     This error is raised when a single broker ``Connection`` goes bad.
#     """
#     def __init__(self, host, port, broker_id=None):
#         self.host = host
#         self.port = port
#         self.broker_id = broker_id
# 
#     def __str__(self):
#         return "Error connecting to %s:%s" % (self.host, self.port)
# 
# Path: kiel/protocol/metadata.py
# class MetadataRequest(Request):
# class Broker(Part):
# class PartitionMetadata(Part):
# class TopicMetadata(Part):
# class MetadataResponse(Response):
# 
# Path: kiel/protocol/coordinator.py
# class GroupCoordinatorRequest(Request):
# class GroupCoordinatorResponse(Response):
# 
# Path: kiel/protocol/produce.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class ProduceRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class ProduceResponse(Response):
# 
# Path: kiel/protocol/fetch.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class FetchRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class FetchResponse(Response):
# 
# Path: kiel/protocol/offset.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class OffsetRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class OffsetResponse(Response):
# 
# Path: kiel/protocol/offset_commit.py
# class PartitionRequest(Part):
# class PartitionV1Request(Part):
# class TopicRequest(Part):
# class TopicV1Request(Part):
# class OffsetCommitV0Request(Request):
# class OffsetCommitV1Request(Request):
# class OffsetCommitV2Request(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class OffsetCommitResponse(Response):
# 
# Path: kiel/protocol/offset_fetch.py
# class TopicRequest(Part):
# class OffsetFetchRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class OffsetFetchResponse(Response):


import contextlib
import logging
import socket
import struct
import sys
from tornado import ioloop, iostream, gen, concurrent
from kiel.exc import BrokerConnectionError
from kiel.protocol import (
    metadata, coordinator,
    produce, fetch,
    offset, offset_commit, offset_fetch
)




log = logging.getLogger(__name__)

# all messages start with a 4-byte signed integer representing raw payload size
size_struct = struct.Struct("!i")
# all responses start with a 4-byte correlation ID to match with the request
correlation_struct = struct.Struct("!i")

response_classes = {
    "metadata": metadata.MetadataResponse,
    "produce": produce.ProduceResponse,
    "fetch": fetch.FetchResponse,
    "offset": offset.OffsetResponse,
    "offset_commit": offset_commit.OffsetCommitResponse,
    "offset_fetch": offset_fetch.OffsetFetchResponse,

# Path: kiel/exc.py
# class BrokerConnectionError(KielError):
#     """
#     This error is raised when a single broker ``Connection`` goes bad.
#     """
#     def __init__(self, host, port, broker_id=None):
#         self.host = host
#         self.port = port
#         self.broker_id = broker_id
# 
#     def __str__(self):
#         return "Error connecting to %s:%s" % (self.host, self.port)
# 
# Path: kiel/protocol/metadata.py
# class MetadataRequest(Request):
# class Broker(Part):
# class PartitionMetadata(Part):
# class TopicMetadata(Part):
# class MetadataResponse(Response):
# 
# Path: kiel/protocol/coordinator.py
# class GroupCoordinatorRequest(Request):
# class GroupCoordinatorResponse(Response):
# 
# Path: kiel/protocol/produce.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class ProduceRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class ProduceResponse(Response):
# 
# Path: kiel/protocol/fetch.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class FetchRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class FetchResponse(Response):
# 
# Path: kiel/protocol/offset.py
# class PartitionRequest(Part):
# class TopicRequest(Part):
# class OffsetRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class OffsetResponse(Response):
# 
# Path: kiel/protocol/offset_commit.py
# class PartitionRequest(Part):
# class PartitionV1Request(Part):
# class TopicRequest(Part):
# class TopicV1Request(Part):
# class OffsetCommitV0Request(Request):
# class OffsetCommitV1Request(Request):
# class OffsetCommitV2Request(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class OffsetCommitResponse(Response):
# 
# Path: kiel/protocol/offset_fetch.py
# class TopicRequest(Part):
# class OffsetFetchRequest(Request):
# class PartitionResponse(Part):
# class TopicResponse(Part):
# class OffsetFetchResponse(Response):


# Path: kiel/connection.py
import contextlib
import logging
import socket
import struct
import sys
from tornado import ioloop, iostream, gen, concurrent
from kiel.exc import BrokerConnectionError
from kiel.protocol import (
    metadata, coordinator,
    produce, fetch,
    offset, offset_commit, offset_fetch
)





log = logging.getLogger(__name__)

# all messages start with a 4-byte signed integer representing raw payload size
size_struct = struct.Struct("!i")
# all responses start with a 4-byte correlation ID to match with the request
correlation_struct = struct.Struct("!i")

response_classes = {
    "metadata": metadata.MetadataResponse,
    "produce": produce.ProduceResponse,
    "fetch": fetch.FetchResponse,
    "offset": offset.OffsetResponse,
    "offset_commit": offset_commit.OffsetCommitResponse,
    "offset_fetch": offset_fetch.OffsetFetchResponse,

    "group_coordinator": coordinator.GroupCoordinatorResponse