# 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
)
        This is never used directly and is fired as a separate callback on the
        I/O loop via the `connect()` method.
        """
        while not self.closing:
            with self.socket_error_handling("Error reading from socket."):
                message = yield self.read_message()
                self.pending.pop(message.correlation_id).set_result(message)

    def abort(self):
        """
        Aborts a connection and puts all pending futures into an error state.

        If ``sys.exc_info()`` is set (i.e. this is being called in an exception
        handler) then pending futures will have that exc info set.  Otherwise
        a ``BrokerConnectionError`` is used.
        """
        if self.closing:
            return

        log.warn("Aborting connection to %s:%s", self.host, self.port)

        self.close()
        self.api_correlation.clear()
        while self.pending:
            _, pending = self.pending.popitem()
            exc_info = sys.exc_info()
            if any(exc_info):
                pending.set_exc_info(sys.exc_info())
            else:
                pending.set_exception(

# 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
)

        This is never used directly and is fired as a separate callback on the
        I/O loop via the `connect()` method.
        """
        while not self.closing:
            with self.socket_error_handling("Error reading from socket."):
                message = yield self.read_message()
                self.pending.pop(message.correlation_id).set_result(message)

    def abort(self):
        """
        Aborts a connection and puts all pending futures into an error state.

        If ``sys.exc_info()`` is set (i.e. this is being called in an exception
        handler) then pending futures will have that exc info set.  Otherwise
        a ``BrokerConnectionError`` is used.
        """
        if self.closing:
            return

        log.warn("Aborting connection to %s:%s", self.host, self.port)

        self.close()
        self.api_correlation.clear()
        while self.pending:
            _, pending = self.pending.popitem()
            exc_info = sys.exc_info()
            if any(exc_info):
                pending.set_exc_info(sys.exc_info())
            else:
                pending.set_exception(

                    BrokerConnectionError(self.host, self.port)