vitaly-krugl: @gmr, please take a look when you have a moment
diff --git a/docs/examples/asynchronous_publisher_example.rst b/docs/examples/asynchronous_publisher_example.rst
index e1267ad..6cba07c 100644
--- a/docs/examples/asynchronous_publisher_example.rst
+++ b/docs/examples/asynchronous_publisher_example.rst
@@ -162,7 +162,7 @@ publisher.py::
             different parameters. In this case, we'll close the connection
             to shutdown the object.
 
-            :param pika.channel.Channel channel: The closed channel
+            :param pika.channel.Channel: The closed channel
             :param int reply_code: The numeric reason the channel was closed
             :param str reply_text: The text reason the channel was closed
 
diff --git a/docs/examples/blocking_consume.rst b/docs/examples/blocking_consume.rst
index 8603c15..85852e4 100644
--- a/docs/examples/blocking_consume.rst
+++ b/docs/examples/blocking_consume.rst
@@ -9,21 +9,21 @@ When pika calls your method, it will pass in the channel, a :py:class:`pika.spec
 
 Example of consuming messages and acknowledging them::
 
-        import pika
+    import pika
 
 
-        def on_message(channel, method_frame, header_frame, body):
-            print(method_frame.delivery_tag)
-	    print(body)
-	    print()
-	    channel.basic_ack(delivery_tag=method_frame.delivery_tag)
+    def on_message(channel, method_frame, header_frame, body):
+        print(method_frame.delivery_tag)
+        print(body)
+        print()
+        channel.basic_ack(delivery_tag=method_frame.delivery_tag)
 
 
-        connection = pika.BlockingConnection()
-        channel = connection.channel()
-        channel.basic_consume(on_message, 'test')
-        try:
-            channel.start_consuming()
-        except KeyboardInterrupt:
-            channel.stop_consuming()
-        connection.close()
+    connection = pika.BlockingConnection()
+    channel = connection.channel()
+    channel.basic_consume(on_message, 'test')
+    try:
+        channel.start_consuming()
+    except KeyboardInterrupt:
+        channel.stop_consuming()
+    connection.close()
\ No newline at end of file
diff --git a/docs/examples/heartbeat_and_blocked_timeouts.rst b/docs/examples/heartbeat_and_blocked_timeouts.rst
new file mode 100644
index 0000000..ba2c4b5
--- /dev/null
+++ b/docs/examples/heartbeat_and_blocked_timeouts.rst
@@ -0,0 +1,37 @@
+Ensuring well-behaved connection with heartbeat and blocked-connection timeouts
+===============================================================================
+
+
+This example demonstrates explicit setting of heartbeat and blocked connection timeouts.
+
+Starting with RabbitMQ 3.5.5, the broker's default hearbeat timeout decreased from 580 seconds to 60 seconds. As a result, applications that perform lengthy processing in the same thread that also runs their Pika connection may experience unexpected dropped connections due to heartbeat timeout. Here, we specify an explicit lower bound for heartbeat timeout.
+
+When RabbitMQ broker is running out of certain resources, such as memory and disk space, it may block connections that are performing resource-consuming operations, such as publishing messages. Once a connection is blocked, RabbiMQ stops reading from that connection's socket, so no commands from the client will get through to te broker on that connection until the broker unblocks it. A blocked connection may last for an indefinite period of time, stalling the connection and possibly resulting in a hang (e.g., in BlockingConnection) until the connection is unblocked. Blocked Connectin Timeout is intended to interrupt (i.e., drop) a connection that has been blocked longer than the given timeout value.
+
+Example of configuring hertbeat and blocked-connection timeouts::
+
+    import pika
+
+
+    def main():
+
+        # NOTE: These paramerers work with all Pika connection types
+        params = pika.ConnectionParameters(heartbeat_interval=600,
+                                           blocked_connection_timeout=300)
+
+        conn = pika.BlockingConnection(params)
+
+        chan = conn.channel()
+
+        chan.basic_publish('', 'my-alphabet-queue', "abc")
+
+        # If publish causes the connection to become blocked, then this conn.close()
+        # would hang until the connection is unblocked, if ever. However, the
+        # blocked_connection_timeout connection parameter would interrupt the wait,
+        # resulting in ConnectionClosed exception from BlockingConnection (or the
+        # on_connection_closed callback call in an asynchronous adapter)
+        conn.close()
+
+
+    if __name__ == '__main__':
+        main()
diff --git a/docs/version_history.rst b/docs/version_history.rst
index c3eb70d..9d7ffc3 100644
--- a/docs/version_history.rst
+++ b/docs/version_history.rst
@@ -18,6 +18,10 @@ Next Release
    never be serviced in the asynchronous scenario.
  - `Channel.basic_reject` fixed to allow `delivery_tag` to be of type `long`
    as well as `int`. (by quantum5)
+ - Implemented support for blocked connection timeouts in
+   `pika.connection.Connection`. This feature is available to all pika adapters.
+   See `pika.connection.ConnectionParameters` docstring to learn more about
+   `blocked_connection_timeout` configuration.
 
 0.10.0 2015-09-02
 -----------------
diff --git a/examples/heatbeat_and_blocked_timeouts.py b/examples/heatbeat_and_blocked_timeouts.py
new file mode 100644
index 0000000..3fe9a99
--- /dev/null
+++ b/examples/heatbeat_and_blocked_timeouts.py
@@ -0,0 +1,48 @@
+"""
+This example demonstrates explicit setting of heartbeat and blocked connection
+timeouts.
+
+Starting with RabbitMQ 3.5.5, the broker's default hearbeat timeout decreased
+from 580 seconds to 60 seconds. As a result, applications that perform lengthy
+processing in the same thread that also runs their Pika connection may
+experience unexpected dropped connections due to heartbeat timeout. Here, we
+specify an explicit lower bound for heartbeat timeout.
+
+When RabbitMQ broker is running out of certain resources, such as memory and
+disk space, it may block connections that are performing resource-consuming
+operations, such as publishing messages. Once a connection is blocked, RabbiMQ
+stops reading from that connection's socket, so no commands from the client will
+get through to te broker on that connection until the broker unblocks it. A
+blocked connection may last for an indefinite period of time, stalling the
+connection and possibly resulting in a hang (e.g., in BlockingConnection) until
+the connection is unblocked. Blocked Connectin Timeout is intended to interrupt
+(i.e., drop) a connection that has been blocked longer than the given timeout
+value.
+"""
+
+
+import pika
+
+
+def main():
+
+    # NOTE: These paramerers work with all Pika connection types
+    params = pika.ConnectionParameters(heartbeat_interval=600,
+                                       blocked_connection_timeout=300)
+
+    conn = pika.BlockingConnection(params)
+
+    chan = conn.channel()
+
+    chan.basic_publish('', 'my-alphabet-queue', "abc")
+
+    # If publish causes the connection to become blocked, then this conn.close()
+    # would hang until the connection is unblocked, if ever. However, the
+    # blocked_connection_timeout connection parameter would interrupt the wait,
+    # resulting in ConnectionClosed exception from BlockingConnection (or the
+    # on_connection_closed callback call in an asynchronous adapter)
+    conn.close()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/pika/adapters/base_connection.py b/pika/adapters/base_connection.py
index d355c60..3bfc71a 100644
--- a/pika/adapters/base_connection.py
+++ b/pika/adapters/base_connection.py
@@ -78,6 +78,29 @@ class BaseConnection(connection.Connection):
                                              on_open_error_callback,
                                              on_close_callback)
 
+    def __repr__(self):
+        def get_socket_repr(sock):
+            if sock is None:
+                return None
+
+            sockname = sock.getsockname()
+
+            peername = None
+            try:
+                peername = sock.getpeername()
+            except socket.error:
+                # not connected?
+                pass
+
+            return '%s->%s' % (sockname, peername)
+
+        return (
+            '<%s state=%s socket=%s params=%s>' %
+            (self.__class__.__name__,
+             self.connection_state,
+             get_socket_repr(self.socket),
+             self.params))
+
     def add_timeout(self, deadline, callback_method):
         """Add the callback_method to the IOLoop timer to fire after deadline
         seconds. Returns a handle to the timeout
@@ -315,7 +338,8 @@ class BaseConnection(connection.Connection):
             LOGGER.error("Socket Error: %s", error_code)
 
         # Disconnect from our IOLoop and let Connection know what's up
-        self._on_terminate(-1, repr(error_value))
+        self._on_terminate(connection.InternalCloseReasons.SOCKET_ERROR,
+                           repr(error_value))
 
     def _handle_timeout(self):
         """Handle a socket timeout in read or write.
@@ -349,7 +373,8 @@ class BaseConnection(connection.Connection):
             error_msg = ('BAD libc:  Write-Only but Read+Error. '
                          'Assume socket disconnected.')
             LOGGER.error(error_msg)
-            self._on_terminate(-1, error_msg)
+            self._on_terminate(connection.InternalCloseReasons.SOCKET_ERROR,
+                               error_msg)
 
         if self.socket and (events & self.ERROR):
             LOGGER.error('Error event %r, %r', events, error)
@@ -391,7 +416,9 @@ class BaseConnection(connection.Connection):
         # Empty data, should disconnect
         if not data or data == 0:
             LOGGER.error('Read empty data, calling disconnect')
-            return self._on_terminate(-1, "EOF")
+            return self._on_terminate(
+                connection.InternalCloseReasons.SOCKET_ERROR,
+                "EOF")
 
         # Pass the data into our top level frame dispatching method
         self._on_data_available(data)
@@ -400,13 +427,13 @@ class BaseConnection(connection.Connection):
     def _handle_write(self):
         """Try and write as much as we can, if we get blocked requeue
         what's left"""
-        bytes_written = 0
+        total_bytes_sent = 0
         try:
             while self.outbound_buffer:
                 frame = self.outbound_buffer.popleft()
                 while True:
                     try:
-                        bw = self.socket.send(frame)
+                        num_bytes_sent = self.socket.send(frame)
                         break
                     except _SOCKET_ERROR as error:
                         if error.errno == errno.EINTR:
@@ -414,10 +441,10 @@ class BaseConnection(connection.Connection):
                         else:
                             raise
 
-                bytes_written += bw
-                if bw < len(frame):
+                total_bytes_sent += num_bytes_sent
+                if num_bytes_sent < len(frame):
                     LOGGER.debug("Partial write, requeing remaining data")
-                    self.outbound_buffer.appendleft(frame[bw:])
+                    self.outbound_buffer.appendleft(frame[num_bytes_sent:])
                     break
 
         except socket.timeout:
@@ -433,7 +460,7 @@ class BaseConnection(connection.Connection):
             else:
                 return self._handle_error(error)
 
-        return bytes_written
+        return total_bytes_sent
 
 
     def _init_connection_state(self):
diff --git a/pika/adapters/blocking_connection.py b/pika/adapters/blocking_connection.py
index d93413b..9c74de5 100644
--- a/pika/adapters/blocking_connection.py
+++ b/pika/adapters/blocking_connection.py
@@ -212,7 +212,7 @@ class _TimerEvt(object):  # pylint: disable=R0903
         self.timer_id = None
 
     def __repr__(self):
-        return '%s(timer_id=%s, callback=%s)' % (self.__class__.__name__,
+        return '<%s timer_id=%s callback=%s>' % (self.__class__.__name__,
                                                  self.timer_id, self._callback)
 
     def dispatch(self):
@@ -236,9 +236,9 @@ class _ConnectionBlockedUnblockedEvtBase(object):  # pylint: disable=R0903
         self._method_frame = method_frame
 
     def __repr__(self):
-        return '%s(callback=%s, frame=%s)' % (self.__class__.__name__,
-                                              self._callback,
-                                              self._method_frame)
+        return '<%s callback=%s, frame=%s>' % (self.__class__.__name__,
+                                               self._callback,
+                                               self._method_frame)
 
     def dispatch(self):
         """Dispatch the user's callback method"""
@@ -266,7 +266,7 @@ class BlockingConnection(object):  # pylint: disable=R0902
     receive messages from RabbitMQ using
     :meth:`basic_consume <BlockingChannel.basic_consume>` or if you want to be
     notified of a delivery failure when using
-    :meth:`basic_publish <BlockingChannel.basic_publish>` .
+    :meth:`basic_publish <BlockingChannel.basic_publish>`.
 
     For more information about communicating with the blocking_connection
     adapter, be sure to check out the
@@ -274,6 +274,40 @@ class BlockingConnection(object):  # pylint: disable=R0902
     :class:`Channel <pika.channel.Channel>` based communication for the
     blocking_connection adapter.
 
+    To prevent recursion/reentrancy, the blocking connection and channel
+    implementations queue asynchronously-delivered events received
+    in nested context (e.g., while waiting for `BlockingConnection.channel` or
+    `BlockingChannel.queue_declare` to complete), dispatching them synchronously
+    once nesting returns to the desired context. This concerns all callbacks,
+    such as those registered via `BlockingConnection.add_timeout`,
+    `BlockingConnection.add_on_connection_blocked_callback`,
+    `BlockingConnection.add_on_connection_unblocked_callback`,
+    `BlockingChannel.basic_consume`, etc.
+
+    Blocked Connection deadlock avoidance: when RabbitMQ becomes low on
+    resources, it emits Connection.Blocked (AMQP extension) to the client
+    connection when client makes a resource-consuming request on that connection
+    or its channel (e.g., `Basic.Publish`); subsequently, RabbitMQ suspsends
+    processing requests from that connection until the affected resources are
+    restored. See http://www.rabbitmq.com/connection-blocked.html. This
+    may impact `BlockingConnection` and `BlockingChannel` operations in a
+    way that users might not be expecting. For example, if the user dispatches
+    `BlockingChannel.basic_publish` in non-publisher-confirmation mode while
+    RabbitMQ is in this low-resource state followed by a synchronous request
+    (e.g., `BlockingConnection.channel`, `BlockingChannel.consume`,
+    `BlockingChannel.basic_consume`, etc.), the synchronous request will block
+    indefinitely (until Connection.Unblocked) waiting for RabbitMQ to reply. If
+    the blocked state persists for a long time, the blocking operation will
+    appear to hang. In this state, `BlockingConnection` instance and its
+    channels will not dispatch user callbacks. SOLUTION: To break this potential
+    deadlock, applications may configure the `blocked_connection_timeout`
+    connection parameter when instantiating `BlockingConnection`. Upon blocked
+    connection timeout, this adapter will raise ConnectionClosed exception with
+    first exception arg of
+    `pika.connection.InternalCloseReasons.BLOCKED_CONNECTION_TIMEOUT`. See
+    `pika.connection.ConnectionParameters` documentation to learn more about
+    `blocked_connection_timeout` configuration.
+
     """
     # Connection-opened callback args
     _OnOpenedArgs = namedtuple('BlockingConnection__OnOpenedArgs',
@@ -341,6 +375,9 @@ class BlockingConnection(object):  # pylint: disable=R0902
 
         self._process_io_for_connection_setup()
 
+    def __repr__(self):
+        return '<%s impl=%r>' % (self.__class__.__name__, self._impl)
+
     def _cleanup(self):
         """Clean up members that might inhibit garbage collection"""
         self._impl.ioloop.deactivate_poller()
@@ -541,8 +578,10 @@ class BlockingConnection(object):  # pylint: disable=R0902
         instead of relying on back pressure throttling. The callback
         will be passed the `Connection.Blocked` method frame.
 
+        See also `ConnectionParameters.blocked_connection_timeout`.
+
         :param method callback_method: Callback to call on `Connection.Blocked`,
-            having the signature callback_method(pika.frame.Method), where the
+            having the signature `callback_method(pika.frame.Method)`, where the
             method frame's `method` member is of type
             `pika.spec.Connection.Blocked`
 
@@ -559,7 +598,7 @@ class BlockingConnection(object):  # pylint: disable=R0902
 
         :param method callback_method: Callback to call on
             `Connection.Unblocked`, having the signature
-            callback_method(pika.frame.Method), where the method frame's
+            `callback_method(pika.frame.Method)`, where the method frame's
             `method` member is of type `pika.spec.Connection.Unblocked`
 
         """
@@ -640,7 +679,12 @@ class BlockingConnection(object):  # pylint: disable=R0902
         for impl_channel in pika.compat.dictvalues(self._impl._channels):
             channel = impl_channel._get_cookie()
             if channel.is_open:
-                channel.close(reply_code, reply_text)
+                try:
+                    channel.close(reply_code, reply_text)
+                except exceptions.ChannelClosed as exc:
+                    # Log and suppress broker-closed channel
+                    LOGGER.warning('Got ChannelClosed while closing channel '
+                                   'from connection.close: %r', exc)
 
         # Close the connection
         self._impl.close(reply_code, reply_text)
@@ -844,8 +888,8 @@ class _ConsumerCancellationEvt(_ChannelPendingEvt):  # pylint: disable=R0903
         self.method_frame = method_frame
 
     def __repr__(self):
-        return '%s(method_frame=%r)' % (self.__class__.__name__,
-                                        self.method_frame)
+        return '<%s method_frame=%r>' % (self.__class__.__name__,
+                                         self.method_frame)
 
     @property
     def method(self):
@@ -879,10 +923,10 @@ class _ReturnedMessageEvt(_ChannelPendingEvt):  # pylint: disable=R0903
         self.body = body
 
     def __repr__(self):
-        return ('%s(callback=%r, channel=%r, method=%r, properties=%r, '
-                'body=%.300r') % (self.__class__.__name__, self.callback,
-                                  self.channel, self.method, self.properties,
-                                  self.body)
+        return ('<%s callback=%r channel=%r method=%r properties=%r '
+                'body=%.300r>') % (self.__class__.__name__, self.callback,
+                                   self.channel, self.method, self.properties,
+                                   self.body)
 
     def dispatch(self):
         """Dispatch user's callback"""
@@ -989,7 +1033,7 @@ class _QueueConsumerGeneratorInfo(object):  # pylint: disable=R0903
         self.pending_events = deque()
 
     def __repr__(self):
-        return '%s(params=%r, consumer_tag=%r)' % (
+        return '<%s params=%r consumer_tag=%r>' % (
             self.__class__.__name__, self.params, self.consumer_tag)
 
 
@@ -1118,20 +1162,26 @@ class BlockingChannel(object):  # pylint: disable=R0904,R0902
 
         LOGGER.info("Created channel=%s", self.channel_number)
 
-    def _cleanup(self):
-        """Clean up members that might inhibit garbage collection"""
-        self._message_confirmation_result.reset()
-        self._pending_events = deque()
-        self._consumer_infos = dict()
-
     def __int__(self):
         """Return the channel object as its channel number
 
+        NOTE: inherited from legacy BlockingConnection; might be error-prone;
+        use `channel_number` property instead.
+
         :rtype: int
 
         """
         return self.channel_number
 
+    def __repr__(self):
+        return '<%s impl=%r>' % (self.__class__.__name__, self._impl)
+
+    def _cleanup(self):
+        """Clean up members that might inhibit garbage collection"""
+        self._message_confirmation_result.reset()
+        self._pending_events = deque()
+        self._consumer_infos = dict()
+
     @property
     def channel_number(self):
         """Channel number"""
diff --git a/pika/adapters/select_connection.py b/pika/adapters/select_connection.py
index 875c48a..ca7b53d 100644
--- a/pika/adapters/select_connection.py
+++ b/pika/adapters/select_connection.py
@@ -167,7 +167,7 @@ class IOLoop(object):
         return self._poller.add_timeout(deadline, callback_method)
 
     def remove_timeout(self, timeout_id):
-        """[API] Remove a timeout if it's still in the timeout stack
+        """[API] Remove a timeout
 
         :param str timeout_id: The timeout id to remove
 
diff --git a/pika/adapters/twisted_connection.py b/pika/adapters/twisted_connection.py
index 62e595c..e2c7625 100644
--- a/pika/adapters/twisted_connection.py
+++ b/pika/adapters/twisted_connection.py
@@ -16,6 +16,7 @@ import functools
 from twisted.internet import defer, error, reactor
 from twisted.python import log
 
+from pika import connection
 from pika import exceptions
 from pika.adapters import base_connection
 
@@ -338,7 +339,8 @@ class TwistedConnection(base_connection.BaseConnection):
         if not reason.check(error.ConnectionDone):
             log.err(reason)
 
-        self._on_terminate(-1, str(reason))
+        self._on_terminate(connection.InternalCloseReasons.SOCKET_ERROR,
+                           str(reason))
 
     def doRead(self):
         self._handle_read()
diff --git a/pika/channel.py b/pika/channel.py
index b7afe54..2165479 100644
--- a/pika/channel.py
+++ b/pika/channel.py
@@ -75,6 +75,11 @@ class Channel(object):
         """
         return self.channel_number
 
+    def __repr__(self):
+        return '<%s number=%s conn=%r>' % (self.__class__.__name__,
+                                           self.channel_number,
+                                           self.connection)
+
     def add_callback(self, callback, replies, one_shot=True):
         """Pass in a callback handler and a list replies from the
         RabbitMQ broker which you'd like the callback notified of. Callbacks
@@ -943,9 +948,10 @@ class Channel(object):
 
         """
         LOGGER.info('%s', method_frame)
-        LOGGER.warning('Received remote Channel.Close (%s): %s',
+        LOGGER.warning('Received remote Channel.Close (%s): %r on channel %s',
                        method_frame.method.reply_code,
-                       method_frame.method.reply_text)
+                       method_frame.method.reply_text,
+                       self)
         if self.connection.is_open:
             self._send_method(spec.Channel.CloseOk())
         self._set_state(self.CLOSED)
diff --git a/pika/connection.py b/pika/connection.py
index 51b0b44..1bbf8e7 100644
--- a/pika/connection.py
+++ b/pika/connection.py
@@ -35,6 +35,16 @@ PRODUCT = "Pika Python Client Library"
 LOGGER = logging.getLogger(__name__)
 
 
+class InternalCloseReasons(object):
+    """Internal reason codes passed to the user's on_close_callback.
+
+    AMQP 0.9.1 specification sites IETF RFC 821 for reply codes. To avoid
+    conflict, the `InternalCloseReasons` namespace uses negative integers.
+    """
+    SOCKET_ERROR = -1
+    BLOCKED_CONNECTION_TIMEOUT = -2
+
+
 class Parameters(object):
     """Base connection parameters class definition
 
@@ -54,6 +64,7 @@ class Parameters(object):
     :param dict DEFAULT_SSL_OPTIONS: {}
     :param int DEFAULT_SSL_PORT: 5671
     :param bool DEFAULT_BACKPRESSURE_DETECTION: False
+    :param number DEFAULT_BLOCKED_CONNECTION_TIMEOUT: None
 
     """
     DEFAULT_BACKPRESSURE_DETECTION = False
@@ -72,6 +83,7 @@ class Parameters(object):
     DEFAULT_SSL_PORT = 5671
     DEFAULT_USERNAME = 'guest'
     DEFAULT_VIRTUAL_HOST = '/'
+    DEFAULT_BLOCKED_CONNECTION_TIMEOUT = None
 
     def __init__(self):
         self.virtual_host = self.DEFAULT_VIRTUAL_HOST
@@ -90,6 +102,13 @@ class Parameters(object):
         self.ssl_options = self.DEFAULT_SSL_OPTIONS
         self.socket_timeout = self.DEFAULT_SOCKET_TIMEOUT
 
+        # If not None, blocked_connection_timeout is the timeout, in seconds,
+        # for the connection to remain blocked; if the timeout expires, the
+        # connection will be torn down, triggering the connection's
+        # on_close_callback
+        self.blocked_connection_timeout = (
+            self.DEFAULT_BLOCKED_CONNECTION_TIMEOUT)
+
     def __repr__(self):
         """Represent the info about the instance.
 
@@ -276,6 +295,23 @@ class Parameters(object):
             raise ValueError('socket_timeout must be > 0')
         return True
 
+    @staticmethod
+    def _validate_blocked_connection_timeout(blocked_connection_timeout):
+        """Validate that the blocked_connection_timeout value is None or a
+        number
+
+        :param real blocked_connection_timeout: The value to validate
+        :rtype: bool
+        :raises: TypeError
+
+        """
+        if blocked_connection_timeout is not None:
+            if not isinstance(blocked_connection_timeout, (int, float)):
+                raise TypeError('blocked_connection_timeout must be a Real number')
+            if blocked_connection_timeout < 0:
+                raise ValueError('blocked_connection_timeout must be >= 0')
+        return True
+
     @staticmethod
     def _validate_ssl(ssl):
         """Validate the SSL toggle is a bool
@@ -320,21 +356,6 @@ class ConnectionParameters(Parameters):
     """Connection parameters object that is passed into the connection adapter
     upon construction.
 
-    :param str host: Hostname or IP Address to connect to
-    :param int port: TCP port to connect to
-    :param str virtual_host: RabbitMQ virtual host to use
-    :param pika.credentials.Credentials credentials: auth credentials
-    :param int channel_max: Maximum number of channels to allow
-    :param int frame_max: The maximum byte size for an AMQP frame
-    :param int heartbeat_interval: How often to send heartbeats
-    :param bool ssl: Enable SSL
-    :param dict ssl_options: Arguments passed to ssl.wrap_socket as
-    :param int connection_attempts: Maximum number of retry attempts
-    :param int|float retry_delay: Time to wait in seconds, before the next
-    :param int|float socket_timeout: Use for high latency networks
-    :param str locale: Set the locale value
-    :param bool backpressure_detection: Toggle backpressure detection
-
     """
 
     def __init__(self,
@@ -351,7 +372,8 @@ class ConnectionParameters(Parameters):
                  retry_delay=None,
                  socket_timeout=None,
                  locale=None,
-                 backpressure_detection=None):
+                 backpressure_detection=None,
+                 blocked_connection_timeout=None):
         """Create a new ConnectionParameters instance.
 
         :param str host: Hostname or IP Address to connect to
@@ -371,6 +393,15 @@ class ConnectionParameters(Parameters):
         :param int|float socket_timeout: Use for high latency networks
         :param str locale: Set the locale value
         :param bool backpressure_detection: Toggle backpressure detection
+        :param blocked_connection_timeout: If not None,
+            the value is a non-negative timeout, in seconds, for the
+            connection to remain blocked (triggered by Connection.Blocked from
+            broker); if the timeout expires before connection becomes unblocked,
+            the connection will be torn down, triggering the adapter-specific
+            mechanism for informing client app about the closed connection (
+            e.g., on_close_callback or ConnectionClosed exception) with
+            `reason_code` of `InternalCloseReasons.BLOCKED_CONNECTION_TIMEOUT`.
+        :type blocked_connection_timeout: None, int, float
 
         """
         super(ConnectionParameters, self).__init__()
@@ -413,6 +444,9 @@ class ConnectionParameters(Parameters):
         if (backpressure_detection is not None and
                 self._validate_backpressure(backpressure_detection)):
             self.backpressure_detection = backpressure_detection
+        if self._validate_blocked_connection_timeout(
+                blocked_connection_timeout):
+            self.blocked_connection_timeout = blocked_connection_timeout
 
 
 class URLParameters(Parameters):
@@ -447,6 +481,11 @@ class URLParameters(Parameters):
             connection failure.
         - socket_timeout:
             Override low level socket timeout value
+        - blocked_connection_timeout:
+            Set the timeout, in seconds, that the connection may remain blocked
+            (triggered by Connection.Blocked from broker); if the timeout
+            expires before connection becomes unblocked, the connection will be
+            torn down, triggering the connection's on_close_callback
 
     :param str url: The AMQP URL to connect to
 
@@ -540,8 +579,8 @@ class URLParameters(Parameters):
             self.heartbeat = values['heartbeat']
 
         elif ('heartbeat_interval' in values and
-                self._validate_heartbeat_interval(
-                    values['heartbeat_interval'])):
+              self._validate_heartbeat_interval(
+                  values['heartbeat_interval'])):
             warnings.warn('heartbeat_interval is deprecated, use heartbeat',
                           DeprecationWarning, stacklevel=2)
             self.heartbeat = values['heartbeat_interval']
@@ -557,6 +596,12 @@ class URLParameters(Parameters):
                 self._validate_socket_timeout(values['socket_timeout'])):
             self.socket_timeout = values['socket_timeout']
 
+        if ('blocked_connection_timeout' in values and
+                self._validate_blocked_connection_timeout(
+                    values['blocked_connection_timeout'])):
+            self.blocked_connection_timeout = values[
+                'blocked_connection_timeout']
+
         if 'ssl_options' in values:
             options = ast.literal_eval(values['ssl_options'])
             if self._validate_ssl_options(options):
@@ -606,21 +651,35 @@ class Connection(object):
         :param method on_open_error_callback: Called if the connection can't
             be established: on_open_error_callback(connection, str|exception)
         :param method on_close_callback: Called when the connection is closed:
-            on_close_callback(connection, reason_code, reason_text)
+            `on_close_callback(connection, reason_code, reason_text)`, where
+            `reason_code` is either an IETF RFC 821 reply code for AMQP-level
+          closures or a value from `pika.connection.InternalCloseReasons` for
+          internal causes, such as socket errors.
 
         """
         self._write_lock = threading.Lock()
 
+        self.connection_state = self.CONNECTION_CLOSED
+
+        # Used to hold timer if configured for Connection.Blocked timeout
+        self._blocked_conn_timer = None
+
+        self.heartbeat = None
+
+        # Set our configuration options
+        self.params = parameters or ConnectionParameters()
+
         # Define our callback dictionary
         self.callbacks = callback.CallbackManager()
 
+        # Initialize the connection state and connect
+        self._init_connection_state()
+
         # Add the on connection error callback
         self.callbacks.add(0, self.ON_CONNECTION_ERROR,
                            on_open_error_callback or self._on_connection_error,
                            False)
 
-        self.heartbeat = None
-
         # On connection callback
         if on_open_callback:
             self.add_on_open_callback(on_open_callback)
@@ -628,12 +687,6 @@ class Connection(object):
         # On connection callback
         if on_close_callback:
             self.add_on_close_callback(on_close_callback)
-
-        # Set our configuration options
-        self.params = parameters or ConnectionParameters()
-
-        # Initialize the connection state and connect
-        self._init_connection_state()
         self.connect()
 
     def add_backpressure_callback(self, callback_method):
@@ -663,7 +716,12 @@ class Connection(object):
         instead of relying on back pressure throttling. The callback
         will be passed the ``Connection.Blocked`` method frame.
 
-        :param method callback_method: Callback to call on `Connection.Blocked`
+        See also `ConnectionParameters.blocked_connection_timeout`.
+
+        :param method callback_method: Callback to call on `Connection.Blocked`,
+            having the signature `callback_method(pika.frame.Method)`, where the
+            method frame's `method` member is of type
+            `pika.spec.Connection.Blocked`
 
         """
         self.callbacks.add(0, spec.Connection.Blocked, callback_method, False)
@@ -675,7 +733,9 @@ class Connection(object):
         ``Connection.Unblocked`` method frame.
 
         :param method callback_method: Callback to call on
-                                       `Connection.Unblocked`
+            `Connection.Unblocked`, having the signature
+            `callback_method(pika.frame.Method)`, where the method frame's
+            `method` member is of type `pika.spec.Connection.Unblocked`
 
         """
         self.callbacks.add(0, spec.Connection.Unblocked, callback_method, False)
@@ -782,12 +842,10 @@ class Connection(object):
             self.remaining_connection_attempts = self.params.connection_attempts
             self._set_connection_state(self.CONNECTION_CLOSED)
 
-    def remove_timeout(self, callback_method):
-        """Adapters should override to call the callback after the
-        specified number of seconds have elapsed, using a timer, or a
-        thread, or similar.
+    def remove_timeout(self, timeout_id):
+        """Adapters should override: Remove a timeout
 
-        :param method callback_method: The callback to remove a timeout for
+        :param str timeout_id: The timeout id to remove
 
         """
         raise NotImplementedError
@@ -941,7 +999,7 @@ class Connection(object):
 
         """
         if (value.method.version_major,
-            value.method.version_minor) != spec.PROTOCOL_VERSION[0:2]:
+                value.method.version_minor) != spec.PROTOCOL_VERSION[0:2]:
             raise exceptions.ProtocolVersionMismatch(frame.ProtocolHeader(),
                                                      value)
 
@@ -1171,6 +1229,18 @@ class Connection(object):
         # simply closed the TCP/IP stream.
         self.callbacks.add(0, spec.Connection.Close, self._on_connection_close)
 
+        if self.params.blocked_connection_timeout is not None:
+            if self._blocked_conn_timer is not None:
+                # Blocked connection timer was active when teardown was
+                # initiated
+                self.remove_timeout(self._blocked_conn_timer)
+                self._blocked_conn_timer = None
+
+            self.add_on_connection_blocked_callback(
+                self._on_connection_blocked)
+            self.add_on_connection_unblocked_callback(
+                self._on_connection_unblocked)
+
     def _is_basic_deliver_frame(self, frame_value):
         """Returns true if the frame is a Basic.Deliver
 
@@ -1250,6 +1320,51 @@ class Connection(object):
         # Start the communication with the RabbitMQ Broker
         self._send_frame(frame.ProtocolHeader())
 
+    def _on_blocked_connection_timeout(self):
+        """ Called when the "connection blocked timeout" expires. When this
+        happens, we tear down the connection
+
+        """
+        self._blocked_conn_timer = None
+        self._on_terminate(InternalCloseReasons.BLOCKED_CONNECTION_TIMEOUT,
+                           'Blocked connection timeout expired')
+
+    def _on_connection_blocked(self, method_frame):
+        """Handle Connection.Blocked notification from RabbitMQ broker
+
+        :param pika.frame.Method method_frame: method frame having `method`
+            member of type `pika.spec.Connection.Blocked`
+        """
+        LOGGER.warning('Received %s from broker', method_frame)
+
+        if self._blocked_conn_timer is not None:
+            # RabbitMQ is not supposed to repeat Connection.Blocked, but it
+            # doesn't hurt to be careful
+            LOGGER.warning('_blocked_conn_timer %s already set when '
+                           '_on_connection_blocked is called',
+                           self._blocked_conn_timer)
+        else:
+            self._blocked_conn_timer = self.add_timeout(
+                self.params.blocked_connection_timeout,
+                self._on_blocked_connection_timeout)
+
+    def _on_connection_unblocked(self, method_frame):
+        """Handle Connection.Unblocked notification from RabbitMQ broker
+
+        :param pika.frame.Method method_frame: method frame having `method`
+            member of type `pika.spec.Connection.Blocked`
+        """
+        LOGGER.info('Received %s from broker', method_frame)
+
+        if self._blocked_conn_timer is None:
+            # RabbitMQ is supposed to pair Connection.Blocked/Unblocked, but it
+            # doesn't hurt to be careful
+            LOGGER.warning('_blocked_conn_timer was not active when '
+                           '_on_connection_unblocked called')
+        else:
+            self.remove_timeout(self._blocked_conn_timer)
+            self._blocked_conn_timer = None
+
     def _on_connection_close(self, method_frame):
         """Called when the connection is closed remotely via Connection.Close
         frame from broker.
@@ -1412,8 +1527,9 @@ class Connection(object):
         """Terminate the connection and notify registered ON_CONNECTION_ERROR
         and/or ON_CONNECTION_CLOSED callbacks
 
-        :param integer reason_code: HTTP error code for AMQP-reported closures
-          or -1 for other errors (such as socket errors)
+        :param integer reason_code: either IETF RFC 821 reply code for
+            AMQP-level closures or a value from `InternalCloseReasons` for
+            internal causes, such as socket errors
         :param str reason_text: human-readable text message describing the error
         """
         LOGGER.warning(
@@ -1436,6 +1552,10 @@ class Connection(object):
         self._remove_callbacks(0, [spec.Connection.Close, spec.Connection.Start,
                                    spec.Connection.Open])
 
+        if self.params.blocked_connection_timeout is not None:
+            self._remove_callbacks(0, [spec.Connection.Blocked,
+                                       spec.Connection.Unblocked])
+
         # Close the socket
         self._adapter_disconnect()
 
@@ -1501,7 +1621,7 @@ class Connection(object):
 
         """
         if (self._is_method_frame(frame_value) and
-            self._has_pending_callbacks(frame_value)):
+                self._has_pending_callbacks(frame_value)):
             self.callbacks.process(frame_value.channel_number,  # Prefix
                                    frame_value.method,  # Key
                                    self,  # Caller

Connection blocked events do not get processed when calling `consume()`
If RabbitMQ is below a resource limit when I connect, the BlockingConnection can end up in a hanging state if I first publish a message and then directly go into `consume()`, even when using the `inactivity_timeout` parameter.

As far as I can see the problem seems to be that when `consume()` sends the `basic_consume` message it waits for the `OK` from Rabbit and does not call the defined `on_blocked_connection_callback` before entering the actual consumption loop. Obviously I could wait for a second after publishing my message and then call `process_data_events()` before entering `consume()`, but that's not a solution. 

How to reproduce:

- Make sure you have a RabbitMQ which is already below resource limits. For example by setting `disk_free_limit` to some really high value.

- This callback gets called as long as the `time.sleep(1)` is there and the sleep is longer than it takes for the `blocked_connection` event to arrive. If you remove the sleep, then it just hangs and the callback does not get called because it's waiting for Rabbit's `OK` message before entering the consumption loop:

        import time
        import pika

        connection_parameters = pika.ConnectionParameters(
            host='192.168.1.101',
            port=5672,
            credentials=pika.PlainCredentials('guest', 'guest'),
        )

        conn = pika.BlockingConnection(connection_parameters)


        def callback(method):
            print('connection blocked!')

        conn.add_on_connection_blocked_callback(callback)

        chan = conn.channel()

        chan.basic_publish(
            exchange='test_exchange',
            routing_key='key',
            body='test'
        )

        time.sleep(1)

        conn.process_data_events()

        for msg in chan.consume(queue='somequeue', inactivity_timeout=1):
            print(msg)

A possible solution might be to also make this `_flush_output` handle the `connection_blocked` event: https://github.com/pika/pika/blob/master/pika/adapters/blocking_connection.py#L1522
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index 9a44208..39446b8 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -15,6 +15,9 @@ import uuid
 
 from pika import spec
 from pika.compat import as_bytes
+import pika.connection
+import pika.frame
+import pika.spec
 
 from async_test_base import (AsyncTestCase, BoundQueueTestCase, AsyncAdapters)
 
@@ -390,3 +393,68 @@ class TestZ_AccessDenied(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
     def on_open(self, connection):
         super(TestZ_AccessDenied, self).on_open(connection)
         self.stop()
+
+
+class TestBlockedConnectionTimesOut(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
+    DESCRIPTION = "Verify that blocked connection terminates on timeout"
+
+    def start(self, *args, **kwargs):
+        self.parameters.blocked_connection_timeout = 0.001
+        self.on_closed_pair = None
+        super(TestBlockedConnectionTimesOut, self).start(*args, **kwargs)
+        self.assertEqual(
+            self.on_closed_pair,
+            (pika.connection.InternalCloseReasons.BLOCKED_CONNECTION_TIMEOUT,
+             'Blocked connection timeout expired'))
+
+    def begin(self, channel):
+
+        # Simulate Connection.Blocked
+        channel.connection._on_connection_blocked(pika.frame.Method(
+            0,
+            pika.spec.Connection.Blocked('Testing blocked connection timeout')))
+
+    def on_closed(self, connection, reply_code, reply_text):
+        """called when the connection has finished closing"""
+        self.on_closed_pair = (reply_code, reply_text)
+        super(TestBlockedConnectionTimesOut, self).on_closed(connection,
+                                                             reply_code,
+                                                             reply_text)
+
+
+class TestBlockedConnectionUnblocks(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
+    DESCRIPTION = "Verify that blocked-unblocked connection closes normally"
+
+    def start(self, *args, **kwargs):
+        self.parameters.blocked_connection_timeout = 0.001
+        self.on_closed_pair = None
+        super(TestBlockedConnectionUnblocks, self).start(*args, **kwargs)
+        self.assertEqual(
+            self.on_closed_pair,
+            (200, 'Normal shutdown'))
+
+    def begin(self, channel):
+
+        # Simulate Connection.Blocked
+        channel.connection._on_connection_blocked(pika.frame.Method(
+            0,
+            pika.spec.Connection.Blocked(
+                'Testing blocked connection unblocks')))
+
+        # Simulate Connection.Unblocked
+        channel.connection._on_connection_unblocked(pika.frame.Method(
+            0,
+            pika.spec.Connection.Unblocked()))
+
+        # Schedule shutdown after blocked connection timeout would expire
+        channel.connection.add_timeout(0.005, self.on_cleanup_timer)
+
+    def on_cleanup_timer(self):
+        self.stop()
+
+    def on_closed(self, connection, reply_code, reply_text):
+        """called when the connection has finished closing"""
+        self.on_closed_pair = (reply_code, reply_text)
+        super(TestBlockedConnectionUnblocks, self).on_closed(connection,
+                                                             reply_code,
+                                                             reply_text)
diff --git a/tests/acceptance/blocking_adapter_test.py b/tests/acceptance/blocking_adapter_test.py
index a6e781a..4fac0f9 100644
--- a/tests/acceptance/blocking_adapter_test.py
+++ b/tests/acceptance/blocking_adapter_test.py
@@ -376,7 +376,7 @@ class TestProcessDataEvents(BlockingTestCaseBase):
         self.assertLess(elapsed, 0.25)
 
 
-class TestConnectionBlockAndUnblock(BlockingTestCaseBase):
+class TestConnectionRegisterForBlockAndUnblock(BlockingTestCaseBase):
 
     def test(self):
         """BlockingConnection register for Connection.Blocked/Unblocked"""
@@ -407,6 +407,35 @@ class TestConnectionBlockAndUnblock(BlockingTestCaseBase):
         self.assertEqual(unblocked_buffer, ["unblocked"])
 
 
+class TestBlockedConnectionTimeout(BlockingTestCaseBase):
+
+    def test(self):
+        """BlockingConnection Connection.Blocked timeout """
+        url = DEFAULT_URL + '&blocked_connection_timeout=0.001'
+        conn = self._connect(url=url)
+
+        # NOTE: I haven't figured out yet how to coerce RabbitMQ to emit
+        # Connection.Block and Connection.Unblock from the test, so we'll
+        # simulate it for now
+
+        # Simulate Connection.Blocked
+        conn._impl._on_connection_blocked(pika.frame.Method(
+            0,
+            pika.spec.Connection.Blocked('TestBlockedConnectionTimeout')))
+
+        # Wait for connection teardown
+        with self.assertRaises(pika.exceptions.ConnectionClosed) as excCtx:
+            while True:
+                conn.process_data_events(time_limit=1)
+
+        self.assertEqual(
+            excCtx.exception.args,
+            (pika.connection.InternalCloseReasons.BLOCKED_CONNECTION_TIMEOUT,
+             'Blocked connection timeout expired'))
+
+
+
+
 class TestAddTimeoutRemoveTimeout(BlockingTestCaseBase):
 
     def test(self):
diff --git a/tests/unit/blocking_connection_tests.py b/tests/unit/blocking_connection_tests.py
index 8814bc9..775e490 100644
--- a/tests/unit/blocking_connection_tests.py
+++ b/tests/unit/blocking_connection_tests.py
@@ -18,7 +18,6 @@ Tests for pika.adapters.blocking_connection.BlockingConnection
 
 import socket
 
-from pika.exceptions import AMQPConnectionError
 
 try:
     from unittest import mock  # pylint: disable=E0611
@@ -33,6 +32,8 @@ except ImportError:
 
 import pika
 from pika.adapters import blocking_connection
+import pika.channel
+from pika.exceptions import AMQPConnectionError, ChannelClosed
 
 
 class BlockingConnectionMockTemplate(blocking_connection.BlockingConnection):
@@ -198,7 +199,8 @@ class BlockingConnectionTests(unittest.TestCase):
                                '_process_io_for_connection_setup'):
             connection = blocking_connection.BlockingConnection('params')
 
-        connection._impl._channels = {1: mock.Mock()}
+        impl_channel_mock = mock.Mock()
+        connection._impl._channels = {1: impl_channel_mock}
 
         with mock.patch.object(
                 blocking_connection.BlockingConnection,
@@ -207,6 +209,53 @@ class BlockingConnectionTests(unittest.TestCase):
             connection._closed_result.signal_once()
             connection.close(200, 'text')
 
+        impl_channel_mock._get_cookie.return_value.close.assert_called_once_with(
+            200, 'text')
+        select_connection_class_mock.return_value.close.assert_called_once_with(
+            200, 'text')
+
+    @patch.object(blocking_connection, 'SelectConnection',
+                  spec_set=SelectConnectionTemplate)
+    def test_close_with_channel_closed_exception(self,
+                                                 select_connection_class_mock):
+        select_connection_class_mock.return_value.is_closed = False
+
+        with mock.patch.object(blocking_connection.BlockingConnection,
+                               '_process_io_for_connection_setup'):
+            connection = blocking_connection.BlockingConnection('params')
+
+        channel1_mock = mock.Mock(
+            is_open=True,
+            close=mock.Mock(side_effect=ChannelClosed,
+                            spec_set=pika.channel.Channel.close),
+            spec_set=blocking_connection.BlockingChannel)
+
+        channel2_mock = mock.Mock(
+            is_open=True,
+            spec_set=blocking_connection.BlockingChannel)
+
+        connection._impl._channels = {
+            1: mock.Mock(
+                _get_cookie=mock.Mock(
+                    return_value=channel1_mock,
+                    spec_set=pika.channel.Channel._get_cookie),
+                spec_set=pika.channel.Channel),
+            2: mock.Mock(
+                _get_cookie=mock.Mock(
+                    return_value=channel2_mock,
+                    spec_set=pika.channel.Channel._get_cookie),
+                spec_set=pika.channel.Channel)
+        }
+
+        with mock.patch.object(
+                blocking_connection.BlockingConnection,
+                '_flush_output',
+                spec_set=blocking_connection.BlockingConnection._flush_output):
+            connection._closed_result.signal_once()
+            connection.close(200, 'text')
+
+            channel1_mock.close.assert_called_once_with(200, 'text')
+            channel2_mock.close.assert_called_once_with(200, 'text')
         select_connection_class_mock.return_value.close.assert_called_once_with(
             200, 'text')
 
diff --git a/tests/unit/channel_tests.py b/tests/unit/channel_tests.py
index 93294a3..4a31e89 100644
--- a/tests/unit/channel_tests.py
+++ b/tests/unit/channel_tests.py
@@ -1287,6 +1287,8 @@ class ChannelTests(unittest.TestCase):
         method_frame = frame.Method(self.obj.channel_number,
                                     spec.Channel.Close(999, 'Test_Value'))
         self.obj._on_close(method_frame)
-        warning.assert_called_with('Received remote Channel.Close (%s): %s',
-                                   method_frame.method.reply_code,
-                                   method_frame.method.reply_text)
+        warning.assert_called_with(
+            'Received remote Channel.Close (%s): %r on channel %s',
+            method_frame.method.reply_code,
+            method_frame.method.reply_text,
+            self.obj)
diff --git a/tests/unit/connection_tests.py b/tests/unit/connection_tests.py
index f7e2a53..821dfa3 100644
--- a/tests/unit/connection_tests.py
+++ b/tests/unit/connection_tests.py
@@ -42,14 +42,15 @@ def callback_method():
 
 class ConnectionTests(unittest.TestCase):
 
-    @mock.patch('pika.connection.Connection.connect')
-    def setUp(self, connect):
-        self.connection = connection.Connection()
+    def setUp(self):
+        with mock.patch('pika.connection.Connection.connect'):
+            self.connection = connection.Connection()
+            self.connection._set_connection_state(
+                connection.Connection.CONNECTION_OPEN)
+
         self.channel = mock.Mock(spec=channel.Channel)
         self.channel.is_open = True
         self.connection._channels[1] = self.channel
-        self.connection._set_connection_state(
-            connection.Connection.CONNECTION_OPEN)
 
     def tearDown(self):
         del self.connection
@@ -335,7 +336,8 @@ class ConnectionTests(unittest.TestCase):
             'ssl': True,
             'connection_attempts': 2,
             'locale': 'en',
-            'ssl_options': {'ssl': 'options'}
+            'ssl_options': {'ssl': 'options'},
+            'blocked_connection_timeout': 10.5
         }
         conn = connection.ConnectionParameters(**kwargs)
         #check values
@@ -356,9 +358,10 @@ class ConnectionTests(unittest.TestCase):
             'frame_max': 40000,
             'heartbeat_interval': 7,
             'backpressure_detection': False,
-            'ssl': True
+            'ssl': True,
+            'blocked_connection_timeout': 10.5
         }
-        #Test Type Errors
+        # Test Type Errors
         for bad_field, bad_value in (
                 ('host', 15672), ('port', '5672'), ('virtual_host', True),
                 ('channel_max', '4'), ('frame_max', '5'),
@@ -366,9 +369,13 @@ class ConnectionTests(unittest.TestCase):
                 ('heartbeat_interval', '6'), ('socket_timeout', '42'),
                 ('retry_delay', 'two'), ('backpressure_detection', 'true'),
                 ('ssl', {'ssl': 'dict'}), ('ssl_options', True),
-                ('connection_attempts', 'hello')):
+                ('connection_attempts', 'hello'),
+                ('blocked_connection_timeout', set())):
+
             bkwargs = copy.deepcopy(kwargs)
+
             bkwargs[bad_field] = bad_value
+
             self.assertRaises(TypeError, connection.ConnectionParameters,
                               **bkwargs)
 
@@ -590,3 +597,205 @@ class ConnectionTests(unittest.TestCase):
             self.assertEqual(1, self.connection.frames_received)
             if frame_type == frame.Heartbeat:
                 self.assertTrue(self.connection.heartbeat.received.called)
+
+    @mock.patch.object(connection.Connection, 'connect',
+                       spec_set=connection.Connection.connect)
+    @mock.patch.object(connection.Connection,
+                       'add_on_connection_blocked_callback')
+    @mock.patch.object(connection.Connection,
+                       'add_on_connection_unblocked_callback')
+    def test_create_with_blocked_connection_timeout_config(
+            self,
+            add_on_unblocked_callback_mock,
+            add_on_blocked_callback_mock,
+            connect_mock):
+
+        conn = connection.Connection(
+            parameters=connection.ConnectionParameters(
+                blocked_connection_timeout=60))
+
+        # Check
+        conn.add_on_connection_blocked_callback.assert_called_once_with(
+            conn._on_connection_blocked)
+
+        conn.add_on_connection_unblocked_callback.assert_called_once_with(
+            conn._on_connection_unblocked)
+
+    @mock.patch.object(connection.Connection, 'add_timeout')
+    @mock.patch.object(connection.Connection, 'connect',
+                       spec_set=connection.Connection.connect)
+    def test_connection_blocked_sets_timer(
+            self,
+            connect_mock,
+            add_timeout_mock):
+
+        conn = connection.Connection(
+            parameters=connection.ConnectionParameters(
+                blocked_connection_timeout=60))
+
+        conn._on_connection_blocked(
+            mock.Mock(name='frame.Method(Connection.Blocked)'))
+
+        # Check
+        conn.add_timeout.assert_called_once_with(
+            60,
+            conn._on_blocked_connection_timeout)
+
+        self.assertIsNotNone(conn._blocked_conn_timer)
+
+    @mock.patch.object(connection.Connection, 'add_timeout')
+    @mock.patch.object(connection.Connection, 'connect',
+                       spec_set=connection.Connection.connect)
+    def test_multi_connection_blocked_in_a_row_sets_timer_once(
+            self,
+            connect_mock,
+            add_timeout_mock):
+
+        conn = connection.Connection(
+            parameters=connection.ConnectionParameters(
+                blocked_connection_timeout=60))
+
+        # Simulate Connection.Blocked trigger
+        conn._on_connection_blocked(
+            mock.Mock(name='frame.Method(Connection.Blocked)'))
+
+        # Check
+        conn.add_timeout.assert_called_once_with(
+            60,
+            conn._on_blocked_connection_timeout)
+
+        self.assertIsNotNone(conn._blocked_conn_timer)
+
+        timer = conn._blocked_conn_timer
+
+        # Simulate Connection.Blocked trigger again
+        conn._on_connection_blocked(
+            mock.Mock(name='frame.Method(Connection.Blocked)'))
+
+        self.assertEqual(conn.add_timeout.call_count, 1)
+        self.assertIs(conn._blocked_conn_timer, timer)
+
+    @mock.patch.object(connection.Connection, '_on_terminate')
+    @mock.patch.object(connection.Connection, 'add_timeout',
+                       spec_set=connection.Connection.add_timeout)
+    @mock.patch.object(connection.Connection, 'connect',
+                       spec_set=connection.Connection.connect)
+    def test_blocked_connection_timeout_teminates_connection(
+            self,
+            connect_mock,
+            add_timeout_mock,
+            on_terminate_mock):
+
+        conn = connection.Connection(
+            parameters=connection.ConnectionParameters(
+                blocked_connection_timeout=60))
+
+        conn._on_connection_blocked(
+            mock.Mock(name='frame.Method(Connection.Blocked)'))
+
+        conn._on_blocked_connection_timeout()
+
+        # Check
+        conn._on_terminate.assert_called_once_with(
+            connection.InternalCloseReasons.BLOCKED_CONNECTION_TIMEOUT,
+            'Blocked connection timeout expired')
+
+        self.assertIsNone(conn._blocked_conn_timer)
+
+    @mock.patch.object(connection.Connection, 'remove_timeout')
+    @mock.patch.object(connection.Connection, 'add_timeout',
+                       spec_set=connection.Connection.add_timeout)
+    @mock.patch.object(connection.Connection, 'connect',
+                       spec_set=connection.Connection.connect)
+    def test_connection_unblocked_removes_timer(
+            self,
+            connect_mock,
+            add_timeout_mock,
+            remove_timeout_mock):
+
+        conn = connection.Connection(
+            parameters=connection.ConnectionParameters(
+                blocked_connection_timeout=60))
+
+        conn._on_connection_blocked(
+            mock.Mock(name='frame.Method(Connection.Blocked)'))
+
+        self.assertIsNotNone(conn._blocked_conn_timer)
+
+        timer = conn._blocked_conn_timer
+
+        conn._on_connection_unblocked(
+            mock.Mock(name='frame.Method(Connection.Unblocked)'))
+
+        # Check
+        conn.remove_timeout.assert_called_once_with(timer)
+        self.assertIsNone(conn._blocked_conn_timer)
+
+    @mock.patch.object(connection.Connection, 'remove_timeout')
+    @mock.patch.object(connection.Connection, 'add_timeout',
+                       spec_set=connection.Connection.add_timeout)
+    @mock.patch.object(connection.Connection, 'connect',
+                       spec_set=connection.Connection.connect)
+    def test_multi_connection_unblocked_in_a_row_removes_timer_once(
+            self,
+            connect_mock,
+            add_timeout_mock,
+            remove_timeout_mock):
+
+        conn = connection.Connection(
+            parameters=connection.ConnectionParameters(
+                blocked_connection_timeout=60))
+
+        # Simulate Connection.Blocked
+        conn._on_connection_blocked(
+            mock.Mock(name='frame.Method(Connection.Blocked)'))
+
+        self.assertIsNotNone(conn._blocked_conn_timer)
+
+        timer = conn._blocked_conn_timer
+
+        # Simulate Connection.Unblocked
+        conn._on_connection_unblocked(
+            mock.Mock(name='frame.Method(Connection.Unblocked)'))
+
+        # Check
+        conn.remove_timeout.assert_called_once_with(timer)
+        self.assertIsNone(conn._blocked_conn_timer)
+
+        # Simulate Connection.Unblocked again
+        conn._on_connection_unblocked(
+            mock.Mock(name='frame.Method(Connection.Unblocked)'))
+
+        self.assertEqual(conn.remove_timeout.call_count, 1)
+        self.assertIsNone(conn._blocked_conn_timer)
+
+    @mock.patch.object(connection.Connection, 'remove_timeout')
+    @mock.patch.object(connection.Connection, 'add_timeout',
+                       spec_set=connection.Connection.add_timeout)
+    @mock.patch.object(connection.Connection, 'connect',
+                       spec_set=connection.Connection.connect)
+    @mock.patch.object(connection.Connection, '_adapter_disconnect',
+                       spec_set=connection.Connection._adapter_disconnect)
+    def test_on_terminate_removes_timer(
+            self,
+            adapter_disconnect_mock,
+            connect_mock,
+            add_timeout_mock,
+            remove_timeout_mock):
+
+        conn = connection.Connection(
+            parameters=connection.ConnectionParameters(
+                blocked_connection_timeout=60))
+
+        conn._on_connection_blocked(
+            mock.Mock(name='frame.Method(Connection.Blocked)'))
+
+        self.assertIsNotNone(conn._blocked_conn_timer)
+
+        timer = conn._blocked_conn_timer
+
+        conn._on_terminate(0, 'test_on_terminate_removes_timer')
+
+        # Check
+        conn.remove_timeout.assert_called_once_with(timer)
+        self.assertIsNone(conn._blocked_conn_timer)

pip install -e .[dev]
parse_log_pytest
requirements.txt
yapf pytest
apt-get update apt-get install -y gcc libev-dev
3.5
test-requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs==22.2.0
certifi==2021.5.30
charset-normalizer==2.0.12
codecov==2.1.13
coverage==6.2
idna==3.10
importlib-metadata==4.8.3
iniconfig==1.1.1
mock==5.2.0
nose==1.3.7
packaging==21.3
-e git+https://github.com/pika/pika.git@c467ad22fb8f2fd2bc925fa59d3b082fd035302b#egg=pika
pluggy==1.0.0
py==1.11.0
pyparsing==3.1.4
pytest==7.0.1
requests==2.27.1
tomli==1.2.3
tornado==6.1
Twisted==15.3.0
typing_extensions==4.1.1
urllib3==1.26.20
yapf==0.32.0
zipp==3.6.0
zope.interface==5.5.2

name: pika
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - attrs==22.2.0
      - charset-normalizer==2.0.12
      - codecov==2.1.13
      - coverage==6.2
      - idna==3.10
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - mock==5.2.0
      - nose==1.3.7
      - packaging==21.3
      - pluggy==1.0.0
      - py==1.11.0
      - pyparsing==3.1.4
      - pytest==7.0.1
      - requests==2.27.1
      - tomli==1.2.3
      - tornado==6.1
      - twisted==15.3.0
      - typing-extensions==4.1.1
      - urllib3==1.26.20
      - yapf==0.32.0
      - zipp==3.6.0
      - zope-interface==5.5.2
prefix: /opt/conda/envs/pika

tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_close_with_channel_closed_exception tests/unit/channel_tests.py::ChannelTests::test_on_close_warning tests/unit/connection_tests.py::ConnectionTests::test_blocked_connection_timeout_teminates_connection tests/unit/connection_tests.py::ConnectionTests::test_connection_blocked_sets_timer tests/unit/connection_tests.py::ConnectionTests::test_connection_unblocked_removes_timer tests/unit/connection_tests.py::ConnectionTests::test_create_with_blocked_connection_timeout_config tests/unit/connection_tests.py::ConnectionTests::test_good_connection_parameters tests/unit/connection_tests.py::ConnectionTests::test_multi_connection_blocked_in_a_row_sets_timer_once tests/unit/connection_tests.py::ConnectionTests::test_multi_connection_unblocked_in_a_row_removes_timer_once tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_removes_timer
tests/acceptance/blocking_adapter_test.py::TestCreateAndCloseConnection::test tests/acceptance/blocking_adapter_test.py::TestMultiCloseConnection::test tests/acceptance/blocking_adapter_test.py::TestConnectionContextManagerClosesConnection::test tests/acceptance/blocking_adapter_test.py::TestConnectionContextManagerClosesConnectionAndPassesOriginalException::test tests/acceptance/blocking_adapter_test.py::TestConnectionContextManagerClosesConnectionAndPassesSystemException::test tests/acceptance/blocking_adapter_test.py::TestInvalidExchangeTypeRaisesConnectionClosed::test tests/acceptance/blocking_adapter_test.py::TestCreateAndCloseConnectionWithChannelAndConsumer::test tests/acceptance/blocking_adapter_test.py::TestSuddenBrokerDisconnectBeforeChannel::test tests/acceptance/blocking_adapter_test.py::TestNoAccessToFileDescriptorAfterConnectionClosed::test tests/acceptance/blocking_adapter_test.py::TestDisconnectDuringConnectionStart::test tests/acceptance/blocking_adapter_test.py::TestDisconnectDuringConnectionTune::test tests/acceptance/blocking_adapter_test.py::TestProcessDataEvents::test tests/acceptance/blocking_adapter_test.py::TestConnectionRegisterForBlockAndUnblock::test tests/acceptance/blocking_adapter_test.py::TestBlockedConnectionTimeout::test tests/acceptance/blocking_adapter_test.py::TestAddTimeoutRemoveTimeout::test tests/acceptance/blocking_adapter_test.py::TestRemoveTimeoutFromTimeoutCallback::test tests/acceptance/blocking_adapter_test.py::TestSleep::test tests/acceptance/blocking_adapter_test.py::TestConnectionProperties::test tests/acceptance/blocking_adapter_test.py::TestCreateAndCloseChannel::test tests/acceptance/blocking_adapter_test.py::TestExchangeDeclareAndDelete::test tests/acceptance/blocking_adapter_test.py::TestExchangeBindAndUnbind::test tests/acceptance/blocking_adapter_test.py::TestQueueDeclareAndDelete::test tests/acceptance/blocking_adapter_test.py::TestPassiveQueueDeclareOfUnknownQueueRaisesChannelClosed::test tests/acceptance/blocking_adapter_test.py::TestQueueBindAndUnbindAndPurge::test tests/acceptance/blocking_adapter_test.py::TestBasicGet::test tests/acceptance/blocking_adapter_test.py::TestBasicReject::test tests/acceptance/blocking_adapter_test.py::TestBasicRejectNoRequeue::test tests/acceptance/blocking_adapter_test.py::TestBasicNack::test tests/acceptance/blocking_adapter_test.py::TestBasicNackNoRequeue::test tests/acceptance/blocking_adapter_test.py::TestBasicNackMultiple::test tests/acceptance/blocking_adapter_test.py::TestBasicRecoverWithRequeue::test tests/acceptance/blocking_adapter_test.py::TestTxCommit::test tests/acceptance/blocking_adapter_test.py::TestTxRollback::test tests/acceptance/blocking_adapter_test.py::TestBasicConsumeFromUnknownQueueRaisesChannelClosed::test tests/acceptance/blocking_adapter_test.py::TestPublishAndBasicPublishWithPubacksUnroutable::test tests/acceptance/blocking_adapter_test.py::TestConfirmDeliveryAfterUnroutableMessage::test tests/acceptance/blocking_adapter_test.py::TestUnroutableMessagesReturnedInNonPubackMode::test tests/acceptance/blocking_adapter_test.py::TestUnroutableMessageReturnedInPubackMode::test tests/acceptance/blocking_adapter_test.py::TestBasicPublishDeliveredWhenPendingUnroutable::test tests/acceptance/blocking_adapter_test.py::TestPublishAndConsumeWithPubacksAndQosOfOne::test tests/acceptance/blocking_adapter_test.py::TestTwoBasicConsumersOnSameChannel::test tests/acceptance/blocking_adapter_test.py::TestBasicCancelPurgesPendingConsumerCancellationEvt::test tests/acceptance/blocking_adapter_test.py::TestBasicPublishWithoutPubacks::test tests/acceptance/blocking_adapter_test.py::TestPublishFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestStopConsumingFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestCloseChannelFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestCloseConnectionFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestNonPubAckPublishAndConsumeHugeMessage::test tests/acceptance/blocking_adapter_test.py::TestNonPubackPublishAndConsumeManyMessages::test tests/acceptance/blocking_adapter_test.py::TestBasicCancelWithNonAckableConsumer::test tests/acceptance/blocking_adapter_test.py::TestBasicCancelWithAckableConsumer::test tests/acceptance/blocking_adapter_test.py::TestUnackedMessageAutoRestoredToQueueOnChannelClose::test tests/acceptance/blocking_adapter_test.py::TestNoAckMessageNotRestoredToQueueOnChannelClose::test tests/acceptance/blocking_adapter_test.py::TestChannelFlow::test
tests/acceptance/blocking_adapter_test.py::TestConnectWithDownedBroker::test tests/acceptance/blocking_adapter_test.py::TestDisconnectDuringConnectionProtocol::test tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_channel tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_connection_attempts_with_timeout tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_constructor tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_server_initiated_error_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_server_initiated_no_error_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_user_initiated_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_process_io_for_connection_setup tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_process_io_for_connection_setup_fails_with_open_error tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_sleep tests/unit/channel_tests.py::ChannelTests::test_add_callback tests/unit/channel_tests.py::ChannelTests::test_add_callback_multiple_replies tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_basic_cancel_empty_added tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_basic_get_empty_added tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_channel_close_added tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_channel_flow_added tests/unit/channel_tests.py::ChannelTests::test_add_on_cancel_callback tests/unit/channel_tests.py::ChannelTests::test_add_on_close_callback tests/unit/channel_tests.py::ChannelTests::test_add_on_flow_callback tests/unit/channel_tests.py::ChannelTests::test_add_on_return_callback tests/unit/channel_tests.py::ChannelTests::test_basic_ack_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_callback_appended tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_calls_validate tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_channel_cancelled_appended tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_no_consumer_tag tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_on_cancel_appended tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_raises_value_error tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_then_close tests/unit/channel_tests.py::ChannelTests::test_basic_consume_calls_validate tests/unit/channel_tests.py::ChannelTests::test_basic_consume_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumer_tag tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumer_tag_cancelled_full tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumer_tag_in_consumers tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumers_callback_value tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumers_pending_list_is_empty tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumers_rpc_called tests/unit/channel_tests.py::ChannelTests::test_basic_consume_duplicate_consumer_tag_raises tests/unit/channel_tests.py::ChannelTests::test_basic_consume_has_pending_list tests/unit/channel_tests.py::ChannelTests::test_basic_get_callback tests/unit/channel_tests.py::ChannelTests::test_basic_get_calls_validate tests/unit/channel_tests.py::ChannelTests::test_basic_get_send_method_called tests/unit/channel_tests.py::ChannelTests::test_basic_nack_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_nack_send_method_request tests/unit/channel_tests.py::ChannelTests::test_basic_publish_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_publish_send_method_request tests/unit/channel_tests.py::ChannelTests::test_basic_qos_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_qos_rpc_request tests/unit/channel_tests.py::ChannelTests::test_basic_recover_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_recover_rpc_request tests/unit/channel_tests.py::ChannelTests::test_basic_reject_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_reject_send_method_request_with_int_tag tests/unit/channel_tests.py::ChannelTests::test_basic_reject_send_method_request_with_long_tag tests/unit/channel_tests.py::ChannelTests::test_basic_reject_spec_with_int_tag tests/unit/channel_tests.py::ChannelTests::test_basic_reject_spec_with_long_tag tests/unit/channel_tests.py::ChannelTests::test_basic_send_method_calls_rpc tests/unit/channel_tests.py::ChannelTests::test_channel_open_add_callbacks_called tests/unit/channel_tests.py::ChannelTests::test_cleanup tests/unit/channel_tests.py::ChannelTests::test_close_basic_cancel_called tests/unit/channel_tests.py::ChannelTests::test_close_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_close_state tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_basic_ack tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_basic_nack tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_with_nowait_raises_value_error tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_without_nowait_selectok tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_no_callback_callback_call_count tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_no_callback_no_basic_ack_callback tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_no_callback_no_basic_nack_callback tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_raises_method_not_implemented_for_confirms tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_raises_method_not_implemented_for_nack tests/unit/channel_tests.py::ChannelTests::test_consumer_tags tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_with_type_arg_assigns_to_exchange_type tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_with_type_arg_raises_deprecation_warning tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_flow_off_rpc_request tests/unit/channel_tests.py::ChannelTests::test_flow_on_flowok_callback tests/unit/channel_tests.py::ChannelTests::test_flow_on_rpc_request tests/unit/channel_tests.py::ChannelTests::test_flow_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_flow_raises_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_get_pending_message tests/unit/channel_tests.py::ChannelTests::test_get_pending_message_item_popped tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_basic_deliver_called tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_basic_get_called tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_basic_return_called tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_method_returns_none tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_sets_header_frame tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_sets_method_frame tests/unit/channel_tests.py::ChannelTests::test_has_content_false tests/unit/channel_tests.py::ChannelTests::test_has_content_true tests/unit/channel_tests.py::ChannelTests::test_immediate_called_logger_warning tests/unit/channel_tests.py::ChannelTests::test_init_blocked tests/unit/channel_tests.py::ChannelTests::test_init_blocking tests/unit/channel_tests.py::ChannelTests::test_init_callbacks tests/unit/channel_tests.py::ChannelTests::test_init_cancelled tests/unit/channel_tests.py::ChannelTests::test_init_channel_number tests/unit/channel_tests.py::ChannelTests::test_init_connection tests/unit/channel_tests.py::ChannelTests::test_init_consumers tests/unit/channel_tests.py::ChannelTests::test_init_frame_dispatcher tests/unit/channel_tests.py::ChannelTests::test_init_has_on_flow_callback tests/unit/channel_tests.py::ChannelTests::test_init_invalid_channel_number tests/unit/channel_tests.py::ChannelTests::test_init_on_flowok_callback tests/unit/channel_tests.py::ChannelTests::test_init_on_getok_callback tests/unit/channel_tests.py::ChannelTests::test_init_on_openok_callback tests/unit/channel_tests.py::ChannelTests::test_init_pending tests/unit/channel_tests.py::ChannelTests::test_init_state tests/unit/channel_tests.py::ChannelTests::test_is_closed_false tests/unit/channel_tests.py::ChannelTests::test_is_closed_true tests/unit/channel_tests.py::ChannelTests::test_is_closing_false tests/unit/channel_tests.py::ChannelTests::test_is_closing_true tests/unit/channel_tests.py::ChannelTests::test_on_cancel_not_appended_cancelled tests/unit/channel_tests.py::ChannelTests::test_on_cancel_removed_consumer tests/unit/channel_tests.py::ChannelTests::test_on_cancelok_removed_consumer tests/unit/channel_tests.py::ChannelTests::test_on_cancelok_removed_pending tests/unit/channel_tests.py::ChannelTests::test_on_confirm_selectok tests/unit/channel_tests.py::ChannelTests::test_on_deliver_callback_called tests/unit/channel_tests.py::ChannelTests::test_on_deliver_pending_callbacks_called tests/unit/channel_tests.py::ChannelTests::test_on_deliver_pending_called tests/unit/channel_tests.py::ChannelTests::test_on_eventok tests/unit/channel_tests.py::ChannelTests::test_on_flow tests/unit/channel_tests.py::ChannelTests::test_on_flow_with_callback tests/unit/channel_tests.py::ChannelTests::test_on_flowok tests/unit/channel_tests.py::ChannelTests::test_on_flowok_callback_reset tests/unit/channel_tests.py::ChannelTests::test_on_flowok_calls_callback tests/unit/channel_tests.py::ChannelTests::test_on_getempty tests/unit/channel_tests.py::ChannelTests::test_on_getok_callback_called tests/unit/channel_tests.py::ChannelTests::test_on_getok_callback_reset tests/unit/channel_tests.py::ChannelTests::test_on_getok_no_callback tests/unit/channel_tests.py::ChannelTests::test_on_openok_callback_called tests/unit/channel_tests.py::ChannelTests::test_on_openok_no_callback tests/unit/channel_tests.py::ChannelTests::test_on_synchronous_complete tests/unit/channel_tests.py::ChannelTests::test_onreturn tests/unit/channel_tests.py::ChannelTests::test_onreturn_warning tests/unit/channel_tests.py::ChannelTests::test_queue_bind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_bind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_bind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_bind_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_declare_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_declare_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_declare_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_declare_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_delete_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_delete_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_delete_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_delete_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_purge_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_purge_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_purge_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_purge_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_unbind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_unbind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_unbind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_rpc_adds_callback tests/unit/channel_tests.py::ChannelTests::test_rpc_enters_blocking_and_adds_on_synchronous_complete tests/unit/channel_tests.py::ChannelTests::test_rpc_not_blocking_and_no_on_synchronous_complete_when_no_replies tests/unit/channel_tests.py::ChannelTests::test_rpc_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_rpc_throws_type_error_with_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_rpc_throws_value_error_with_unacceptable_replies tests/unit/channel_tests.py::ChannelTests::test_rpc_while_blocking_appends_blocked_collection tests/unit/channel_tests.py::ChannelTests::test_send_method tests/unit/channel_tests.py::ChannelTests::test_set_state tests/unit/channel_tests.py::ChannelTests::test_tx_commit_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_tx_commit_rpc_request tests/unit/channel_tests.py::ChannelTests::test_tx_rollback_rpc_request tests/unit/channel_tests.py::ChannelTests::test_tx_select_rpc_request tests/unit/channel_tests.py::ChannelTests::test_validate_channel_and_callback_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_validate_channel_and_callback_raises_value_error_not_callable tests/unit/connection_tests.py::ConnectionTests::test_add_callbacks tests/unit/connection_tests.py::ConnectionTests::test_add_on_close_callback tests/unit/connection_tests.py::ConnectionTests::test_add_on_open_error_callback tests/unit/connection_tests.py::ConnectionTests::test_bad_type_connection_parameters tests/unit/connection_tests.py::ConnectionTests::test_channel tests/unit/connection_tests.py::ConnectionTests::test_client_properties tests/unit/connection_tests.py::ConnectionTests::test_close_channels tests/unit/connection_tests.py::ConnectionTests::test_close_closes_open_channels tests/unit/connection_tests.py::ConnectionTests::test_close_ignores_closed_channels tests/unit/connection_tests.py::ConnectionTests::test_connect tests/unit/connection_tests.py::ConnectionTests::test_connect_reconnect tests/unit/connection_tests.py::ConnectionTests::test_new_conn_should_use_first_channel tests/unit/connection_tests.py::ConnectionTests::test_next_channel_number_returns_lowest_unused tests/unit/connection_tests.py::ConnectionTests::test_on_channel_cleanup_no_open_channels tests/unit/connection_tests.py::ConnectionTests::test_on_channel_cleanup_non_closing_state tests/unit/connection_tests.py::ConnectionTests::test_on_channel_cleanup_open_channels tests/unit/connection_tests.py::ConnectionTests::test_on_close_ready_no_open_channels tests/unit/connection_tests.py::ConnectionTests::test_on_close_ready_open_channels tests/unit/connection_tests.py::ConnectionTests::test_on_connection_close_ok tests/unit/connection_tests.py::ConnectionTests::test_on_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_on_connection_start tests/unit/connection_tests.py::ConnectionTests::test_on_connection_tune tests/unit/connection_tests.py::ConnectionTests::test_on_data_available tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_cleans_up tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_access_denied_on_connection_error_and_closed tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_auth_on_connection_error_and_closed tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_connection_closed_callback tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_protocol_on_connection_error_and_closed tests/unit/connection_tests.py::ConnectionTests::test_process_url tests/unit/connection_tests.py::ConnectionTests::test_set_backpressure_multiplier
docs/examples/heartbeat_and_blocked_timeouts.rst pika/adapters/blocking_connection.py pika/adapters/select_connection.py pika/adapters/twisted_connection.py pika/channel.py pika/connection.py examples/heatbeat_and_blocked_timeouts.py docs/examples/asynchronous_publisher_example.rst pika/adapters/base_connection.py docs/version_history.rst docs/examples/blocking_consume.rst
docs/examples/heartbeat_and_blocked_timeouts.rst pika/adapters/blocking_connection.py pika/adapters/select_connection.py pika/adapters/twisted_connection.py pika/channel.py pika/connection.py examples/heatbeat_and_blocked_timeouts.py docs/examples/asynchronous_publisher_example.rst pika/adapters/base_connection.py docs/version_history.rst docs/examples/blocking_consume.rst