We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
ValueError: Socket not connected
Hello,
the following error happens sometimes when publishing :

```
  File "/foo/bar/lib/python2.7/site-packages/kombu/messaging.py", line 181, in publish
    exchange_name, declare,
  File "/foo/bar/lib/python2.7/site-packages/kombu/connection.py", line 506, in _ensured
    self.collect()
  File "/foo/bar/lib/python2.7/site-packages/kombu/connection.py", line 350, in collect
    gc_transport(self._connection)
  File "/foo/bar/lib/python2.7/site-packages/kombu/transport/librabbitmq.py", line 148, in _collect
    os.close(connection.fileno())
ValueError: Socket not connected

```

kombu==4.1.0

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kombu/transport/librabbitmq.py`
Content:
```
1 """`librabbitmq`_ transport.
2 
3 .. _`librabbitmq`: https://pypi.python.org/librabbitmq/
4 """
5 from __future__ import absolute_import, unicode_literals
6 
7 import os
8 import socket
9 import warnings
10 
11 import librabbitmq as amqp
12 from librabbitmq import ChannelError, ConnectionError
13 
14 from kombu.five import items, values
15 from kombu.utils.amq_manager import get_manager
16 from kombu.utils.text import version_string_as_tuple
17 
18 from . import base
19 from .base import to_rabbitmq_queue_arguments
20 
21 W_VERSION = """
22     librabbitmq version too old to detect RabbitMQ version information
23     so make sure you are using librabbitmq 1.5 when using rabbitmq > 3.3
24 """
25 DEFAULT_PORT = 5672
26 DEFAULT_SSL_PORT = 5671
27 
28 NO_SSL_ERROR = """\
29 ssl not supported by librabbitmq, please use pyamqp:// or stunnel\
30 """
31 
32 
33 class Message(base.Message):
34     """AMQP Message (librabbitmq)."""
35 
36     def __init__(self, channel, props, info, body):
37         super(Message, self).__init__(
38             channel=channel,
39             body=body,
40             delivery_info=info,
41             properties=props,
42             delivery_tag=info.get('delivery_tag'),
43             content_type=props.get('content_type'),
44             content_encoding=props.get('content_encoding'),
45             headers=props.get('headers'))
46 
47 
48 class Channel(amqp.Channel, base.StdChannel):
49     """AMQP Channel (librabbitmq)."""
50 
51     Message = Message
52 
53     def prepare_message(self, body, priority=None,
54                         content_type=None, content_encoding=None,
55                         headers=None, properties=None):
56         """Encapsulate data into a AMQP message."""
57         properties = properties if properties is not None else {}
58         properties.update({'content_type': content_type,
59                            'content_encoding': content_encoding,
60                            'headers': headers,
61                            'priority': priority})
62         return body, properties
63 
64     def prepare_queue_arguments(self, arguments, **kwargs):
65         arguments = to_rabbitmq_queue_arguments(arguments, **kwargs)
66         return {k.encode('utf8'): v for k, v in items(arguments)}
67 
68 
69 class Connection(amqp.Connection):
70     """AMQP Connection (librabbitmq)."""
71 
72     Channel = Channel
73     Message = Message
74 
75 
76 class Transport(base.Transport):
77     """AMQP Transport (librabbitmq)."""
78 
79     Connection = Connection
80 
81     default_port = DEFAULT_PORT
82     default_ssl_port = DEFAULT_SSL_PORT
83 
84     connection_errors = (
85         base.Transport.connection_errors + (
86             ConnectionError, socket.error, IOError, OSError)
87     )
88     channel_errors = (
89         base.Transport.channel_errors + (ChannelError,)
90     )
91     driver_type = 'amqp'
92     driver_name = 'librabbitmq'
93 
94     implements = base.Transport.implements.extend(
95         asynchronous=True,
96         heartbeats=False,
97     )
98 
99     def __init__(self, client, **kwargs):
100         self.client = client
101         self.default_port = kwargs.get('default_port') or self.default_port
102         self.default_ssl_port = (kwargs.get('default_ssl_port') or
103                                  self.default_ssl_port)
104         self.__reader = None
105 
106     def driver_version(self):
107         return amqp.__version__
108 
109     def create_channel(self, connection):
110         return connection.channel()
111 
112     def drain_events(self, connection, **kwargs):
113         return connection.drain_events(**kwargs)
114 
115     def establish_connection(self):
116         """Establish connection to the AMQP broker."""
117         conninfo = self.client
118         for name, default_value in items(self.default_connection_params):
119             if not getattr(conninfo, name, None):
120                 setattr(conninfo, name, default_value)
121         if conninfo.ssl:
122             raise NotImplementedError(NO_SSL_ERROR)
123         opts = dict({
124             'host': conninfo.host,
125             'userid': conninfo.userid,
126             'password': conninfo.password,
127             'virtual_host': conninfo.virtual_host,
128             'login_method': conninfo.login_method,
129             'insist': conninfo.insist,
130             'ssl': conninfo.ssl,
131             'connect_timeout': conninfo.connect_timeout,
132         }, **conninfo.transport_options or {})
133         conn = self.Connection(**opts)
134         conn.client = self.client
135         self.client.drain_events = conn.drain_events
136         return conn
137 
138     def close_connection(self, connection):
139         """Close the AMQP broker connection."""
140         self.client.drain_events = None
141         connection.close()
142 
143     def _collect(self, connection):
144         if connection is not None:
145             for channel in values(connection.channels):
146                 channel.connection = None
147             try:
148                 os.close(connection.fileno())
149             except OSError:
150                 pass
151             connection.channels.clear()
152             connection.callbacks.clear()
153         self.client.drain_events = None
154         self.client = None
155 
156     def verify_connection(self, connection):
157         return connection.connected
158 
159     def register_with_event_loop(self, connection, loop):
160         loop.add_reader(
161             connection.fileno(), self.on_readable, connection, loop,
162         )
163 
164     def get_manager(self, *args, **kwargs):
165         return get_manager(self.client, *args, **kwargs)
166 
167     def qos_semantics_matches_spec(self, connection):
168         try:
169             props = connection.server_properties
170         except AttributeError:
171             warnings.warn(UserWarning(W_VERSION))
172         else:
173             if props.get('product') == 'RabbitMQ':
174                 return version_string_as_tuple(props['version']) < (3, 3)
175         return True
176 
177     @property
178     def default_connection_params(self):
179         return {
180             'userid': 'guest',
181             'password': 'guest',
182             'port': (self.default_ssl_port if self.client.ssl
183                      else self.default_port),
184             'hostname': 'localhost',
185             'login_method': 'AMQPLAIN',
186         }
187
```
--- 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/kombu/transport/librabbitmq.py b/kombu/transport/librabbitmq.py
--- a/kombu/transport/librabbitmq.py
+++ b/kombu/transport/librabbitmq.py
@@ -146,7 +146,7 @@
                 channel.connection = None
             try:
                 os.close(connection.fileno())
-            except OSError:
+            except (OSError, ValueError):
                 pass
             connection.channels.clear()
             connection.callbacks.clear()

{"golden_diff": "diff --git a/kombu/transport/librabbitmq.py b/kombu/transport/librabbitmq.py\n--- a/kombu/transport/librabbitmq.py\n+++ b/kombu/transport/librabbitmq.py\n@@ -146,7 +146,7 @@\n                 channel.connection = None\n             try:\n                 os.close(connection.fileno())\n-            except OSError:\n+            except (OSError, ValueError):\n                 pass\n             connection.channels.clear()\n             connection.callbacks.clear()\n", "issue": "ValueError: Socket not connected\nHello,\r\nthe following error happens sometimes when publishing :\r\n\r\n```\r\n  File \"/foo/bar/lib/python2.7/site-packages/kombu/messaging.py\", line 181, in publish\r\n    exchange_name, declare,\r\n  File \"/foo/bar/lib/python2.7/site-packages/kombu/connection.py\", line 506, in _ensured\r\n    self.collect()\r\n  File \"/foo/bar/lib/python2.7/site-packages/kombu/connection.py\", line 350, in collect\r\n    gc_transport(self._connection)\r\n  File \"/foo/bar/lib/python2.7/site-packages/kombu/transport/librabbitmq.py\", line 148, in _collect\r\n    os.close(connection.fileno())\r\nValueError: Socket not connected\r\n\r\n```\r\n\r\nkombu==4.1.0\n", "before_files": [{"content": "\"\"\"`librabbitmq`_ transport.\n\n.. _`librabbitmq`: https://pypi.python.org/librabbitmq/\n\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport os\nimport socket\nimport warnings\n\nimport librabbitmq as amqp\nfrom librabbitmq import ChannelError, ConnectionError\n\nfrom kombu.five import items, values\nfrom kombu.utils.amq_manager import get_manager\nfrom kombu.utils.text import version_string_as_tuple\n\nfrom . import base\nfrom .base import to_rabbitmq_queue_arguments\n\nW_VERSION = \"\"\"\n    librabbitmq version too old to detect RabbitMQ version information\n    so make sure you are using librabbitmq 1.5 when using rabbitmq > 3.3\n\"\"\"\nDEFAULT_PORT = 5672\nDEFAULT_SSL_PORT = 5671\n\nNO_SSL_ERROR = \"\"\"\\\nssl not supported by librabbitmq, please use pyamqp:// or stunnel\\\n\"\"\"\n\n\nclass Message(base.Message):\n    \"\"\"AMQP Message (librabbitmq).\"\"\"\n\n    def __init__(self, channel, props, info, body):\n        super(Message, self).__init__(\n            channel=channel,\n            body=body,\n            delivery_info=info,\n            properties=props,\n            delivery_tag=info.get('delivery_tag'),\n            content_type=props.get('content_type'),\n            content_encoding=props.get('content_encoding'),\n            headers=props.get('headers'))\n\n\nclass Channel(amqp.Channel, base.StdChannel):\n    \"\"\"AMQP Channel (librabbitmq).\"\"\"\n\n    Message = Message\n\n    def prepare_message(self, body, priority=None,\n                        content_type=None, content_encoding=None,\n                        headers=None, properties=None):\n        \"\"\"Encapsulate data into a AMQP message.\"\"\"\n        properties = properties if properties is not None else {}\n        properties.update({'content_type': content_type,\n                           'content_encoding': content_encoding,\n                           'headers': headers,\n                           'priority': priority})\n        return body, properties\n\n    def prepare_queue_arguments(self, arguments, **kwargs):\n        arguments = to_rabbitmq_queue_arguments(arguments, **kwargs)\n        return {k.encode('utf8'): v for k, v in items(arguments)}\n\n\nclass Connection(amqp.Connection):\n    \"\"\"AMQP Connection (librabbitmq).\"\"\"\n\n    Channel = Channel\n    Message = Message\n\n\nclass Transport(base.Transport):\n    \"\"\"AMQP Transport (librabbitmq).\"\"\"\n\n    Connection = Connection\n\n    default_port = DEFAULT_PORT\n    default_ssl_port = DEFAULT_SSL_PORT\n\n    connection_errors = (\n        base.Transport.connection_errors + (\n            ConnectionError, socket.error, IOError, OSError)\n    )\n    channel_errors = (\n        base.Transport.channel_errors + (ChannelError,)\n    )\n    driver_type = 'amqp'\n    driver_name = 'librabbitmq'\n\n    implements = base.Transport.implements.extend(\n        asynchronous=True,\n        heartbeats=False,\n    )\n\n    def __init__(self, client, **kwargs):\n        self.client = client\n        self.default_port = kwargs.get('default_port') or self.default_port\n        self.default_ssl_port = (kwargs.get('default_ssl_port') or\n                                 self.default_ssl_port)\n        self.__reader = None\n\n    def driver_version(self):\n        return amqp.__version__\n\n    def create_channel(self, connection):\n        return connection.channel()\n\n    def drain_events(self, connection, **kwargs):\n        return connection.drain_events(**kwargs)\n\n    def establish_connection(self):\n        \"\"\"Establish connection to the AMQP broker.\"\"\"\n        conninfo = self.client\n        for name, default_value in items(self.default_connection_params):\n            if not getattr(conninfo, name, None):\n                setattr(conninfo, name, default_value)\n        if conninfo.ssl:\n            raise NotImplementedError(NO_SSL_ERROR)\n        opts = dict({\n            'host': conninfo.host,\n            'userid': conninfo.userid,\n            'password': conninfo.password,\n            'virtual_host': conninfo.virtual_host,\n            'login_method': conninfo.login_method,\n            'insist': conninfo.insist,\n            'ssl': conninfo.ssl,\n            'connect_timeout': conninfo.connect_timeout,\n        }, **conninfo.transport_options or {})\n        conn = self.Connection(**opts)\n        conn.client = self.client\n        self.client.drain_events = conn.drain_events\n        return conn\n\n    def close_connection(self, connection):\n        \"\"\"Close the AMQP broker connection.\"\"\"\n        self.client.drain_events = None\n        connection.close()\n\n    def _collect(self, connection):\n        if connection is not None:\n            for channel in values(connection.channels):\n                channel.connection = None\n            try:\n                os.close(connection.fileno())\n            except OSError:\n                pass\n            connection.channels.clear()\n            connection.callbacks.clear()\n        self.client.drain_events = None\n        self.client = None\n\n    def verify_connection(self, connection):\n        return connection.connected\n\n    def register_with_event_loop(self, connection, loop):\n        loop.add_reader(\n            connection.fileno(), self.on_readable, connection, loop,\n        )\n\n    def get_manager(self, *args, **kwargs):\n        return get_manager(self.client, *args, **kwargs)\n\n    def qos_semantics_matches_spec(self, connection):\n        try:\n            props = connection.server_properties\n        except AttributeError:\n            warnings.warn(UserWarning(W_VERSION))\n        else:\n            if props.get('product') == 'RabbitMQ':\n                return version_string_as_tuple(props['version']) < (3, 3)\n        return True\n\n    @property\n    def default_connection_params(self):\n        return {\n            'userid': 'guest',\n            'password': 'guest',\n            'port': (self.default_ssl_port if self.client.ssl\n                     else self.default_port),\n            'hostname': 'localhost',\n            'login_method': 'AMQPLAIN',\n        }\n", "path": "kombu/transport/librabbitmq.py"}], "after_files": [{"content": "\"\"\"`librabbitmq`_ transport.\n\n.. _`librabbitmq`: https://pypi.python.org/librabbitmq/\n\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport os\nimport socket\nimport warnings\n\nimport librabbitmq as amqp\nfrom librabbitmq import ChannelError, ConnectionError\n\nfrom kombu.five import items, values\nfrom kombu.utils.amq_manager import get_manager\nfrom kombu.utils.text import version_string_as_tuple\n\nfrom . import base\nfrom .base import to_rabbitmq_queue_arguments\n\nW_VERSION = \"\"\"\n    librabbitmq version too old to detect RabbitMQ version information\n    so make sure you are using librabbitmq 1.5 when using rabbitmq > 3.3\n\"\"\"\nDEFAULT_PORT = 5672\nDEFAULT_SSL_PORT = 5671\n\nNO_SSL_ERROR = \"\"\"\\\nssl not supported by librabbitmq, please use pyamqp:// or stunnel\\\n\"\"\"\n\n\nclass Message(base.Message):\n    \"\"\"AMQP Message (librabbitmq).\"\"\"\n\n    def __init__(self, channel, props, info, body):\n        super(Message, self).__init__(\n            channel=channel,\n            body=body,\n            delivery_info=info,\n            properties=props,\n            delivery_tag=info.get('delivery_tag'),\n            content_type=props.get('content_type'),\n            content_encoding=props.get('content_encoding'),\n            headers=props.get('headers'))\n\n\nclass Channel(amqp.Channel, base.StdChannel):\n    \"\"\"AMQP Channel (librabbitmq).\"\"\"\n\n    Message = Message\n\n    def prepare_message(self, body, priority=None,\n                        content_type=None, content_encoding=None,\n                        headers=None, properties=None):\n        \"\"\"Encapsulate data into a AMQP message.\"\"\"\n        properties = properties if properties is not None else {}\n        properties.update({'content_type': content_type,\n                           'content_encoding': content_encoding,\n                           'headers': headers,\n                           'priority': priority})\n        return body, properties\n\n    def prepare_queue_arguments(self, arguments, **kwargs):\n        arguments = to_rabbitmq_queue_arguments(arguments, **kwargs)\n        return {k.encode('utf8'): v for k, v in items(arguments)}\n\n\nclass Connection(amqp.Connection):\n    \"\"\"AMQP Connection (librabbitmq).\"\"\"\n\n    Channel = Channel\n    Message = Message\n\n\nclass Transport(base.Transport):\n    \"\"\"AMQP Transport (librabbitmq).\"\"\"\n\n    Connection = Connection\n\n    default_port = DEFAULT_PORT\n    default_ssl_port = DEFAULT_SSL_PORT\n\n    connection_errors = (\n        base.Transport.connection_errors + (\n            ConnectionError, socket.error, IOError, OSError)\n    )\n    channel_errors = (\n        base.Transport.channel_errors + (ChannelError,)\n    )\n    driver_type = 'amqp'\n    driver_name = 'librabbitmq'\n\n    implements = base.Transport.implements.extend(\n        asynchronous=True,\n        heartbeats=False,\n    )\n\n    def __init__(self, client, **kwargs):\n        self.client = client\n        self.default_port = kwargs.get('default_port') or self.default_port\n        self.default_ssl_port = (kwargs.get('default_ssl_port') or\n                                 self.default_ssl_port)\n        self.__reader = None\n\n    def driver_version(self):\n        return amqp.__version__\n\n    def create_channel(self, connection):\n        return connection.channel()\n\n    def drain_events(self, connection, **kwargs):\n        return connection.drain_events(**kwargs)\n\n    def establish_connection(self):\n        \"\"\"Establish connection to the AMQP broker.\"\"\"\n        conninfo = self.client\n        for name, default_value in items(self.default_connection_params):\n            if not getattr(conninfo, name, None):\n                setattr(conninfo, name, default_value)\n        if conninfo.ssl:\n            raise NotImplementedError(NO_SSL_ERROR)\n        opts = dict({\n            'host': conninfo.host,\n            'userid': conninfo.userid,\n            'password': conninfo.password,\n            'virtual_host': conninfo.virtual_host,\n            'login_method': conninfo.login_method,\n            'insist': conninfo.insist,\n            'ssl': conninfo.ssl,\n            'connect_timeout': conninfo.connect_timeout,\n        }, **conninfo.transport_options or {})\n        conn = self.Connection(**opts)\n        conn.client = self.client\n        self.client.drain_events = conn.drain_events\n        return conn\n\n    def close_connection(self, connection):\n        \"\"\"Close the AMQP broker connection.\"\"\"\n        self.client.drain_events = None\n        connection.close()\n\n    def _collect(self, connection):\n        if connection is not None:\n            for channel in values(connection.channels):\n                channel.connection = None\n            try:\n                os.close(connection.fileno())\n            except (OSError, ValueError):\n                pass\n            connection.channels.clear()\n            connection.callbacks.clear()\n        self.client.drain_events = None\n        self.client = None\n\n    def verify_connection(self, connection):\n        return connection.connected\n\n    def register_with_event_loop(self, connection, loop):\n        loop.add_reader(\n            connection.fileno(), self.on_readable, connection, loop,\n        )\n\n    def get_manager(self, *args, **kwargs):\n        return get_manager(self.client, *args, **kwargs)\n\n    def qos_semantics_matches_spec(self, connection):\n        try:\n            props = connection.server_properties\n        except AttributeError:\n            warnings.warn(UserWarning(W_VERSION))\n        else:\n            if props.get('product') == 'RabbitMQ':\n                return version_string_as_tuple(props['version']) < (3, 3)\n        return True\n\n    @property\n    def default_connection_params(self):\n        return {\n            'userid': 'guest',\n            'password': 'guest',\n            'port': (self.default_ssl_port if self.client.ssl\n                     else self.default_port),\n            'hostname': 'localhost',\n            'login_method': 'AMQPLAIN',\n        }\n", "path": "kombu/transport/librabbitmq.py"}]}