We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Clearer Error Logs when RabbitMQ not reachable
If RabbitMQ is not running, users see logs like this in st2api.log:

```
2018-02-06 17:44:10,549 140082567223888 INFO __init__ [-] Connecting to database "st2" @ "127.0.0.1:27017" as user "None".
2018-02-06 17:44:10,654 140082567223888 ERROR connection_retry_wrapper [-] Connection or channel error identified: [Errno 111] ECONNREFUSED.
Traceback (most recent call last):
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/st2common/transport/connection_retry_wrapper.py", line 115, in run
    channel = connection.channel()
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/connection.py", line 266, in channel
    chan = self.transport.create_channel(self.connection)
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/connection.py", line 802, in connection
    self._connection = self._establish_connection()
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/connection.py", line 757, in _establish_connection
    conn = self.transport.establish_connection()
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/transport/pyamqp.py", line 130, in establish_connection
    conn.connect()
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/amqp/connection.py", line 282, in connect
    self.transport.connect()
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/amqp/transport.py", line 109, in connect
    self._connect(self.host, self.port, self.connect_timeout)
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/amqp/transport.py", line 150, in _connect
    self.sock.connect(sa)
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/eventlet/greenio/base.py", line 256, in connect
    socket_checkerr(fd)
  File "/opt/stackstorm/st2/local/lib/python2.7/site-packages/eventlet/greenio/base.py", line 46, in socket_checkerr
    raise socket.error(err, errno.errorcode[err])
error: [Errno 111] ECONNREFUSED
```

This is confusing for users for two reasons. One is because it is immediately below a "Connecting to database" message. Issue #3990 addresses that. The other issue is that the ERROR message here does not make it obvious that it is a problem with connecting to RabbitMQ. This should be more obvious in the log message. 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `st2common/st2common/transport/connection_retry_wrapper.py`
Content:
```
1 # Licensed to the StackStorm, Inc ('StackStorm') under one or more
2 # contributor license agreements.  See the NOTICE file distributed with
3 # this work for additional information regarding copyright ownership.
4 # The ASF licenses this file to You under the Apache License, Version 2.0
5 # (the "License"); you may not use this file except in compliance with
6 # the License.  You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 from __future__ import absolute_import
17 import eventlet
18 
19 __all__ = ['ConnectionRetryWrapper', 'ClusterRetryContext']
20 
21 
22 class ClusterRetryContext(object):
23     """
24     Stores retry context for cluster retries. It makes certain assumptions
25     on how cluster_size and retry should be determined.
26     """
27     def __init__(self, cluster_size):
28         # No of nodes in a cluster
29         self.cluster_size = cluster_size
30         # No of times to retry in a cluster
31         self.cluster_retry = 2
32         # time to wait between retry in a cluster
33         self.wait_between_cluster = 10
34 
35         # No of nodes attempted. Starts at 1 since the
36         self._nodes_attempted = 1
37 
38     def test_should_stop(self):
39         should_stop = True
40         if self._nodes_attempted > self.cluster_size * self.cluster_retry:
41             return should_stop, -1
42         wait = 0
43         should_stop = False
44         if self._nodes_attempted % self.cluster_size == 0:
45             wait = self.wait_between_cluster
46         self._nodes_attempted += 1
47         return should_stop, wait
48 
49 
50 class ConnectionRetryWrapper(object):
51     """
52     Manages retry of connection and also switching to different nodes in a cluster.
53 
54     :param cluster_size: Size of the cluster.
55     :param logger: logger to use to log moderately useful information.
56 
57     .. code-block:: python
58         # Without ensuring recoverable errors are retried
59         connection_urls = [
60             'amqp://guest:guest@node1:5672',
61             'amqp://guest:guest@node2:5672',
62             'amqp://guest:guest@node3:5672'
63         ]
64         with Connection(connection_urls) as connection:
65             retry_wrapper = ConnectionRetryWrapper(cluster_size=len(connection_urls),
66                                                    logger=my_logger)
67             # wrapped_callback must have signature ``def func(connection, channel)``
68             def wrapped_callback(connection, channel):
69                 pass
70 
71             retry_wrapper.run(connection=connection, wrapped_callback=wrapped_callback)
72 
73         # With ensuring recoverable errors are retried
74         connection_urls = [
75             'amqp://guest:guest@node1:5672',
76             'amqp://guest:guest@node2:5672',
77             'amqp://guest:guest@node3:5672'
78         ]
79         with Connection(connection_urls) as connection:
80             retry_wrapper = ConnectionRetryWrapper(cluster_size=len(connection_urls),
81                                                    logger=my_logger)
82             # wrapped_callback must have signature ``def func(connection, channel)``
83             def wrapped_callback(connection, channel):
84                 kwargs = {...}
85                 # call ensured to correctly deal with recoverable errors.
86                 retry_wrapper.ensured(connection=connection_retry_wrapper,
87                                       obj=my_obj,
88                                       to_ensure_func=my_obj.ensuree,
89                                       **kwargs)
90 
91             retry_wrapper.run(connection=connection, wrapped_callback=wrapped_callback)
92 
93     """
94     def __init__(self, cluster_size, logger):
95         self._retry_context = ClusterRetryContext(cluster_size=cluster_size)
96         self._logger = logger
97 
98     def errback(self, exc, interval):
99         self._logger.error('Rabbitmq connection error: %s', exc.message)
100 
101     def run(self, connection, wrapped_callback):
102         """
103         Run the wrapped_callback in a protective covering of retries and error handling.
104 
105         :param connection: Connection to messaging service
106         :type connection: kombu.connection.Connection
107 
108         :param wrapped_callback: Callback that will be wrapped by all the fine handling in this
109                                  method. Expected signature of callback -
110                                  ``def func(connection, channel)``
111         """
112         should_stop = False
113         channel = None
114         while not should_stop:
115             try:
116                 channel = connection.channel()
117                 wrapped_callback(connection=connection, channel=channel)
118                 should_stop = True
119             except connection.connection_errors + connection.channel_errors as e:
120                 self._logger.exception('Connection or channel error identified: %s.' % (str(e)))
121                 should_stop, wait = self._retry_context.test_should_stop()
122                 # reset channel to None to avoid any channel closing errors. At this point
123                 # in case of an exception there should be no channel but that is better to
124                 # guarantee.
125                 channel = None
126                 # All attempts to re-establish connections have failed. This error needs to
127                 # be notified so raise.
128                 if should_stop:
129                     raise
130                 # -1, 0 and 1+ are handled properly by eventlet.sleep
131                 eventlet.sleep(wait)
132 
133                 connection.close()
134                 # ensure_connection will automatically switch to an alternate. Other connections
135                 # in the pool will be fixed independently. It would be nice to cut-over the
136                 # entire ConnectionPool simultaneously but that would require writing our own
137                 # ConnectionPool. If a server recovers it could happen that the same process
138                 # ends up talking to separate nodes in a cluster.
139                 connection.ensure_connection()
140 
141             except Exception as e:
142                 self._logger.exception('Connections to rabbitmq cannot be re-established: %s',
143                                        e.message)
144                 # Not being able to publish a message could be a significant issue for an app.
145                 raise
146             finally:
147                 if should_stop and channel:
148                     try:
149                         channel.close()
150                     except Exception:
151                         self._logger.warning('Error closing channel.', exc_info=True)
152 
153     def ensured(self, connection, obj, to_ensure_func, **kwargs):
154         """
155         Ensure that recoverable errors are retried a set number of times before giving up.
156 
157         :param connection: Connection to messaging service
158         :type connection: kombu.connection.Connection
159 
160         :param obj: Object whose method is to be ensured. Typically, channel, producer etc. from
161                     the kombu library.
162         :type obj: Must support mixin kombu.abstract.MaybeChannelBound
163         """
164         ensuring_func = connection.ensure(obj, to_ensure_func, errback=self.errback, max_retries=3)
165         ensuring_func(**kwargs)
166
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/st2common/st2common/transport/connection_retry_wrapper.py b/st2common/st2common/transport/connection_retry_wrapper.py
--- a/st2common/st2common/transport/connection_retry_wrapper.py
+++ b/st2common/st2common/transport/connection_retry_wrapper.py
@@ -117,7 +117,7 @@
                 wrapped_callback(connection=connection, channel=channel)
                 should_stop = True
             except connection.connection_errors + connection.channel_errors as e:
-                self._logger.exception('Connection or channel error identified: %s.' % (str(e)))
+                self._logger.exception('RabbitMQ connection or channel error: %s.' % (str(e)))
                 should_stop, wait = self._retry_context.test_should_stop()
                 # reset channel to None to avoid any channel closing errors. At this point
                 # in case of an exception there should be no channel but that is better to

{"golden_diff": "diff --git a/st2common/st2common/transport/connection_retry_wrapper.py b/st2common/st2common/transport/connection_retry_wrapper.py\n--- a/st2common/st2common/transport/connection_retry_wrapper.py\n+++ b/st2common/st2common/transport/connection_retry_wrapper.py\n@@ -117,7 +117,7 @@\n                 wrapped_callback(connection=connection, channel=channel)\n                 should_stop = True\n             except connection.connection_errors + connection.channel_errors as e:\n-                self._logger.exception('Connection or channel error identified: %s.' % (str(e)))\n+                self._logger.exception('RabbitMQ connection or channel error: %s.' % (str(e)))\n                 should_stop, wait = self._retry_context.test_should_stop()\n                 # reset channel to None to avoid any channel closing errors. At this point\n                 # in case of an exception there should be no channel but that is better to\n", "issue": "Clearer Error Logs when RabbitMQ not reachable\nIf RabbitMQ is not running, users see logs like this in st2api.log:\r\n\r\n```\r\n2018-02-06 17:44:10,549 140082567223888 INFO __init__ [-] Connecting to database \"st2\" @ \"127.0.0.1:27017\" as user \"None\".\r\n2018-02-06 17:44:10,654 140082567223888 ERROR connection_retry_wrapper [-] Connection or channel error identified: [Errno 111] ECONNREFUSED.\r\nTraceback (most recent call last):\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/st2common/transport/connection_retry_wrapper.py\", line 115, in run\r\n    channel = connection.channel()\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/connection.py\", line 266, in channel\r\n    chan = self.transport.create_channel(self.connection)\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/connection.py\", line 802, in connection\r\n    self._connection = self._establish_connection()\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/connection.py\", line 757, in _establish_connection\r\n    conn = self.transport.establish_connection()\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/kombu/transport/pyamqp.py\", line 130, in establish_connection\r\n    conn.connect()\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/amqp/connection.py\", line 282, in connect\r\n    self.transport.connect()\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/amqp/transport.py\", line 109, in connect\r\n    self._connect(self.host, self.port, self.connect_timeout)\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/amqp/transport.py\", line 150, in _connect\r\n    self.sock.connect(sa)\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/eventlet/greenio/base.py\", line 256, in connect\r\n    socket_checkerr(fd)\r\n  File \"/opt/stackstorm/st2/local/lib/python2.7/site-packages/eventlet/greenio/base.py\", line 46, in socket_checkerr\r\n    raise socket.error(err, errno.errorcode[err])\r\nerror: [Errno 111] ECONNREFUSED\r\n```\r\n\r\nThis is confusing for users for two reasons. One is because it is immediately below a \"Connecting to database\" message. Issue #3990 addresses that. The other issue is that the ERROR message here does not make it obvious that it is a problem with connecting to RabbitMQ. This should be more obvious in the log message. \n", "before_files": [{"content": "# Licensed to the StackStorm, Inc ('StackStorm') under one or more\n# contributor license agreements.  See the NOTICE file distributed with\n# this work for additional information regarding copyright ownership.\n# The ASF licenses this file to You under the Apache License, Version 2.0\n# (the \"License\"); you may not use this file except in compliance with\n# the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\nimport eventlet\n\n__all__ = ['ConnectionRetryWrapper', 'ClusterRetryContext']\n\n\nclass ClusterRetryContext(object):\n    \"\"\"\n    Stores retry context for cluster retries. It makes certain assumptions\n    on how cluster_size and retry should be determined.\n    \"\"\"\n    def __init__(self, cluster_size):\n        # No of nodes in a cluster\n        self.cluster_size = cluster_size\n        # No of times to retry in a cluster\n        self.cluster_retry = 2\n        # time to wait between retry in a cluster\n        self.wait_between_cluster = 10\n\n        # No of nodes attempted. Starts at 1 since the\n        self._nodes_attempted = 1\n\n    def test_should_stop(self):\n        should_stop = True\n        if self._nodes_attempted > self.cluster_size * self.cluster_retry:\n            return should_stop, -1\n        wait = 0\n        should_stop = False\n        if self._nodes_attempted % self.cluster_size == 0:\n            wait = self.wait_between_cluster\n        self._nodes_attempted += 1\n        return should_stop, wait\n\n\nclass ConnectionRetryWrapper(object):\n    \"\"\"\n    Manages retry of connection and also switching to different nodes in a cluster.\n\n    :param cluster_size: Size of the cluster.\n    :param logger: logger to use to log moderately useful information.\n\n    .. code-block:: python\n        # Without ensuring recoverable errors are retried\n        connection_urls = [\n            'amqp://guest:guest@node1:5672',\n            'amqp://guest:guest@node2:5672',\n            'amqp://guest:guest@node3:5672'\n        ]\n        with Connection(connection_urls) as connection:\n            retry_wrapper = ConnectionRetryWrapper(cluster_size=len(connection_urls),\n                                                   logger=my_logger)\n            # wrapped_callback must have signature ``def func(connection, channel)``\n            def wrapped_callback(connection, channel):\n                pass\n\n            retry_wrapper.run(connection=connection, wrapped_callback=wrapped_callback)\n\n        # With ensuring recoverable errors are retried\n        connection_urls = [\n            'amqp://guest:guest@node1:5672',\n            'amqp://guest:guest@node2:5672',\n            'amqp://guest:guest@node3:5672'\n        ]\n        with Connection(connection_urls) as connection:\n            retry_wrapper = ConnectionRetryWrapper(cluster_size=len(connection_urls),\n                                                   logger=my_logger)\n            # wrapped_callback must have signature ``def func(connection, channel)``\n            def wrapped_callback(connection, channel):\n                kwargs = {...}\n                # call ensured to correctly deal with recoverable errors.\n                retry_wrapper.ensured(connection=connection_retry_wrapper,\n                                      obj=my_obj,\n                                      to_ensure_func=my_obj.ensuree,\n                                      **kwargs)\n\n            retry_wrapper.run(connection=connection, wrapped_callback=wrapped_callback)\n\n    \"\"\"\n    def __init__(self, cluster_size, logger):\n        self._retry_context = ClusterRetryContext(cluster_size=cluster_size)\n        self._logger = logger\n\n    def errback(self, exc, interval):\n        self._logger.error('Rabbitmq connection error: %s', exc.message)\n\n    def run(self, connection, wrapped_callback):\n        \"\"\"\n        Run the wrapped_callback in a protective covering of retries and error handling.\n\n        :param connection: Connection to messaging service\n        :type connection: kombu.connection.Connection\n\n        :param wrapped_callback: Callback that will be wrapped by all the fine handling in this\n                                 method. Expected signature of callback -\n                                 ``def func(connection, channel)``\n        \"\"\"\n        should_stop = False\n        channel = None\n        while not should_stop:\n            try:\n                channel = connection.channel()\n                wrapped_callback(connection=connection, channel=channel)\n                should_stop = True\n            except connection.connection_errors + connection.channel_errors as e:\n                self._logger.exception('Connection or channel error identified: %s.' % (str(e)))\n                should_stop, wait = self._retry_context.test_should_stop()\n                # reset channel to None to avoid any channel closing errors. At this point\n                # in case of an exception there should be no channel but that is better to\n                # guarantee.\n                channel = None\n                # All attempts to re-establish connections have failed. This error needs to\n                # be notified so raise.\n                if should_stop:\n                    raise\n                # -1, 0 and 1+ are handled properly by eventlet.sleep\n                eventlet.sleep(wait)\n\n                connection.close()\n                # ensure_connection will automatically switch to an alternate. Other connections\n                # in the pool will be fixed independently. It would be nice to cut-over the\n                # entire ConnectionPool simultaneously but that would require writing our own\n                # ConnectionPool. If a server recovers it could happen that the same process\n                # ends up talking to separate nodes in a cluster.\n                connection.ensure_connection()\n\n            except Exception as e:\n                self._logger.exception('Connections to rabbitmq cannot be re-established: %s',\n                                       e.message)\n                # Not being able to publish a message could be a significant issue for an app.\n                raise\n            finally:\n                if should_stop and channel:\n                    try:\n                        channel.close()\n                    except Exception:\n                        self._logger.warning('Error closing channel.', exc_info=True)\n\n    def ensured(self, connection, obj, to_ensure_func, **kwargs):\n        \"\"\"\n        Ensure that recoverable errors are retried a set number of times before giving up.\n\n        :param connection: Connection to messaging service\n        :type connection: kombu.connection.Connection\n\n        :param obj: Object whose method is to be ensured. Typically, channel, producer etc. from\n                    the kombu library.\n        :type obj: Must support mixin kombu.abstract.MaybeChannelBound\n        \"\"\"\n        ensuring_func = connection.ensure(obj, to_ensure_func, errback=self.errback, max_retries=3)\n        ensuring_func(**kwargs)\n", "path": "st2common/st2common/transport/connection_retry_wrapper.py"}], "after_files": [{"content": "# Licensed to the StackStorm, Inc ('StackStorm') under one or more\n# contributor license agreements.  See the NOTICE file distributed with\n# this work for additional information regarding copyright ownership.\n# The ASF licenses this file to You under the Apache License, Version 2.0\n# (the \"License\"); you may not use this file except in compliance with\n# the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom __future__ import absolute_import\nimport eventlet\n\n__all__ = ['ConnectionRetryWrapper', 'ClusterRetryContext']\n\n\nclass ClusterRetryContext(object):\n    \"\"\"\n    Stores retry context for cluster retries. It makes certain assumptions\n    on how cluster_size and retry should be determined.\n    \"\"\"\n    def __init__(self, cluster_size):\n        # No of nodes in a cluster\n        self.cluster_size = cluster_size\n        # No of times to retry in a cluster\n        self.cluster_retry = 2\n        # time to wait between retry in a cluster\n        self.wait_between_cluster = 10\n\n        # No of nodes attempted. Starts at 1 since the\n        self._nodes_attempted = 1\n\n    def test_should_stop(self):\n        should_stop = True\n        if self._nodes_attempted > self.cluster_size * self.cluster_retry:\n            return should_stop, -1\n        wait = 0\n        should_stop = False\n        if self._nodes_attempted % self.cluster_size == 0:\n            wait = self.wait_between_cluster\n        self._nodes_attempted += 1\n        return should_stop, wait\n\n\nclass ConnectionRetryWrapper(object):\n    \"\"\"\n    Manages retry of connection and also switching to different nodes in a cluster.\n\n    :param cluster_size: Size of the cluster.\n    :param logger: logger to use to log moderately useful information.\n\n    .. code-block:: python\n        # Without ensuring recoverable errors are retried\n        connection_urls = [\n            'amqp://guest:guest@node1:5672',\n            'amqp://guest:guest@node2:5672',\n            'amqp://guest:guest@node3:5672'\n        ]\n        with Connection(connection_urls) as connection:\n            retry_wrapper = ConnectionRetryWrapper(cluster_size=len(connection_urls),\n                                                   logger=my_logger)\n            # wrapped_callback must have signature ``def func(connection, channel)``\n            def wrapped_callback(connection, channel):\n                pass\n\n            retry_wrapper.run(connection=connection, wrapped_callback=wrapped_callback)\n\n        # With ensuring recoverable errors are retried\n        connection_urls = [\n            'amqp://guest:guest@node1:5672',\n            'amqp://guest:guest@node2:5672',\n            'amqp://guest:guest@node3:5672'\n        ]\n        with Connection(connection_urls) as connection:\n            retry_wrapper = ConnectionRetryWrapper(cluster_size=len(connection_urls),\n                                                   logger=my_logger)\n            # wrapped_callback must have signature ``def func(connection, channel)``\n            def wrapped_callback(connection, channel):\n                kwargs = {...}\n                # call ensured to correctly deal with recoverable errors.\n                retry_wrapper.ensured(connection=connection_retry_wrapper,\n                                      obj=my_obj,\n                                      to_ensure_func=my_obj.ensuree,\n                                      **kwargs)\n\n            retry_wrapper.run(connection=connection, wrapped_callback=wrapped_callback)\n\n    \"\"\"\n    def __init__(self, cluster_size, logger):\n        self._retry_context = ClusterRetryContext(cluster_size=cluster_size)\n        self._logger = logger\n\n    def errback(self, exc, interval):\n        self._logger.error('Rabbitmq connection error: %s', exc.message)\n\n    def run(self, connection, wrapped_callback):\n        \"\"\"\n        Run the wrapped_callback in a protective covering of retries and error handling.\n\n        :param connection: Connection to messaging service\n        :type connection: kombu.connection.Connection\n\n        :param wrapped_callback: Callback that will be wrapped by all the fine handling in this\n                                 method. Expected signature of callback -\n                                 ``def func(connection, channel)``\n        \"\"\"\n        should_stop = False\n        channel = None\n        while not should_stop:\n            try:\n                channel = connection.channel()\n                wrapped_callback(connection=connection, channel=channel)\n                should_stop = True\n            except connection.connection_errors + connection.channel_errors as e:\n                self._logger.exception('RabbitMQ connection or channel error: %s.' % (str(e)))\n                should_stop, wait = self._retry_context.test_should_stop()\n                # reset channel to None to avoid any channel closing errors. At this point\n                # in case of an exception there should be no channel but that is better to\n                # guarantee.\n                channel = None\n                # All attempts to re-establish connections have failed. This error needs to\n                # be notified so raise.\n                if should_stop:\n                    raise\n                # -1, 0 and 1+ are handled properly by eventlet.sleep\n                eventlet.sleep(wait)\n\n                connection.close()\n                # ensure_connection will automatically switch to an alternate. Other connections\n                # in the pool will be fixed independently. It would be nice to cut-over the\n                # entire ConnectionPool simultaneously but that would require writing our own\n                # ConnectionPool. If a server recovers it could happen that the same process\n                # ends up talking to separate nodes in a cluster.\n                connection.ensure_connection()\n\n            except Exception as e:\n                self._logger.exception('Connections to rabbitmq cannot be re-established: %s',\n                                       e.message)\n                # Not being able to publish a message could be a significant issue for an app.\n                raise\n            finally:\n                if should_stop and channel:\n                    try:\n                        channel.close()\n                    except Exception:\n                        self._logger.warning('Error closing channel.', exc_info=True)\n\n    def ensured(self, connection, obj, to_ensure_func, **kwargs):\n        \"\"\"\n        Ensure that recoverable errors are retried a set number of times before giving up.\n\n        :param connection: Connection to messaging service\n        :type connection: kombu.connection.Connection\n\n        :param obj: Object whose method is to be ensured. Typically, channel, producer etc. from\n                    the kombu library.\n        :type obj: Must support mixin kombu.abstract.MaybeChannelBound\n        \"\"\"\n        ensuring_func = connection.ensure(obj, to_ensure_func, errback=self.errback, max_retries=3)\n        ensuring_func(**kwargs)\n", "path": "st2common/st2common/transport/connection_retry_wrapper.py"}]}