Repository: pika/pika

Issue:
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

Discussion:
vitaly-krugl: @gmr, please take a look when you have a moment

Patch:
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


Test patch:
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)