We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Mongodb url is broken in Kombu 3.0.1
Upgrading to version 3.0.1, a mongodb url like mongodb://localhost/wq  fails as the // gets converted to / by some code in transport/mongodb.py  - basically it seems to boil down to this code:-

```
    if '/' in hostname[10:]:
        if not client.userid:
            hostname = hostname.replace('/' + client.virtual_host, '/')
        else:
            hostname = hostname.replace('/' + client.virtual_host, '/' + authdb)
```

by default virtual_host == '/' so basically hostname becomes:-

mongodb:/localhost/wq  instead of mongodb://localhost/wq causing the MongoClient to fail wit a port not an integer error.

right now can work around this by setting the url to mongodb:////localhost/wq but its a kludge.

mongodb transport no longer honors configured port.
The mongodb transport in kombu version 2.1.3 (version according to pypi) has changed the BrokerConnection arguments that it uses. We are passing in hostname, port, and transport in as separate BrokerConnection arguments, but the mongodb transport now expects to get all that information out of hostname, which it assumes is a mongodb:// URI.

For some history, version 1.5.0 changed the parse_url() function so that the port was no longer parsed out of mongodb:// URIs. So we were forced to pass the hostname, port, and transport in as separate BrokerConnection arguments. Now, we are being forced back to passing a mongodb:// URI in as the hostname.


--- 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/mongodb.py`
Content:
```
1 """
2 kombu.transport.mongodb
3 =======================
4 
5 MongoDB transport.
6 
7 :copyright: (c) 2010 - 2013 by Flavio Percoco Premoli.
8 :license: BSD, see LICENSE for more details.
9 
10 """
11 from __future__ import absolute_import
12 
13 import pymongo
14 
15 from pymongo import errors
16 from anyjson import loads, dumps
17 from pymongo import MongoClient
18 
19 from kombu.five import Empty
20 from kombu.syn import _detect_environment
21 from kombu.utils.encoding import bytes_to_str
22 
23 from . import virtual
24 
25 DEFAULT_HOST = '127.0.0.1'
26 DEFAULT_PORT = 27017
27 
28 __author__ = """\
29 Flavio [FlaPer87] Percoco Premoli <flaper87@flaper87.org>;\
30 Scott Lyons <scottalyons@gmail.com>;\
31 """
32 
33 
34 class Channel(virtual.Channel):
35     _client = None
36     supports_fanout = True
37     _fanout_queues = {}
38 
39     def __init__(self, *vargs, **kwargs):
40         super_ = super(Channel, self)
41         super_.__init__(*vargs, **kwargs)
42 
43         self._queue_cursors = {}
44         self._queue_readcounts = {}
45 
46     def _new_queue(self, queue, **kwargs):
47         pass
48 
49     def _get(self, queue):
50         try:
51             if queue in self._fanout_queues:
52                 msg = next(self._queue_cursors[queue])
53                 self._queue_readcounts[queue] += 1
54                 return loads(bytes_to_str(msg['payload']))
55             else:
56                 msg = self.client.command(
57                     'findandmodify', 'messages',
58                     query={'queue': queue},
59                     sort={'_id': pymongo.ASCENDING}, remove=True,
60                 )
61         except errors.OperationFailure as exc:
62             if 'No matching object found' in exc.args[0]:
63                 raise Empty()
64             raise
65         except StopIteration:
66             raise Empty()
67 
68         # as of mongo 2.0 empty results won't raise an error
69         if msg['value'] is None:
70             raise Empty()
71         return loads(bytes_to_str(msg['value']['payload']))
72 
73     def _size(self, queue):
74         if queue in self._fanout_queues:
75             return (self._queue_cursors[queue].count() -
76                     self._queue_readcounts[queue])
77 
78         return self.client.messages.find({'queue': queue}).count()
79 
80     def _put(self, queue, message, **kwargs):
81         self.client.messages.insert({'payload': dumps(message),
82                                      'queue': queue})
83 
84     def _purge(self, queue):
85         size = self._size(queue)
86         if queue in self._fanout_queues:
87             cursor = self._queue_cursors[queue]
88             cursor.rewind()
89             self._queue_cursors[queue] = cursor.skip(cursor.count())
90         else:
91             self.client.messages.remove({'queue': queue})
92         return size
93 
94     def _open(self, scheme='mongodb://'):
95         # See mongodb uri documentation:
96         # http://www.mongodb.org/display/DOCS/Connections
97         client = self.connection.client
98         options = client.transport_options
99         hostname = client.hostname or DEFAULT_HOST
100         dbname = client.virtual_host
101 
102         if dbname in ['/', None]:
103             dbname = "kombu_default"
104         if not hostname.startswith(scheme):
105             hostname = scheme + hostname
106 
107         if not hostname[len(scheme):]:
108             hostname += 'localhost'
109 
110         # XXX What does this do?  [ask]
111         urest = hostname[len(scheme):]
112         if '/' in urest:
113             if not client.userid:
114                 urest = urest.replace('/' + client.virtual_host, '/')
115                 hostname = ''.join([scheme, urest])
116 
117         # At this point we expect the hostname to be something like
118         # (considering replica set form too):
119         #
120         #   mongodb://[username:password@]host1[:port1][,host2[:port2],
121         #   ...[,hostN[:portN]]][/[?options]]
122         options.setdefault('auto_start_request', True)
123         mongoconn = MongoClient(
124             host=hostname, ssl=client.ssl,
125             auto_start_request=options['auto_start_request'],
126             use_greenlets=_detect_environment() != 'default',
127         )
128         database = getattr(mongoconn, dbname)
129 
130         version = mongoconn.server_info()['version']
131         if tuple(map(int, version.split('.')[:2])) < (1, 3):
132             raise NotImplementedError(
133                 'Kombu requires MongoDB version 1.3+ (server is {0})'.format(
134                     version))
135 
136         self.db = database
137         col = database.messages
138         col.ensure_index([('queue', 1), ('_id', 1)], background=True)
139 
140         if 'messages.broadcast' not in database.collection_names():
141             capsize = options.get('capped_queue_size') or 100000
142             database.create_collection('messages.broadcast',
143                                        size=capsize, capped=True)
144 
145         self.bcast = getattr(database, 'messages.broadcast')
146         self.bcast.ensure_index([('queue', 1)])
147 
148         self.routing = getattr(database, 'messages.routing')
149         self.routing.ensure_index([('queue', 1), ('exchange', 1)])
150         return database
151 
152     #TODO: Store a more complete exchange metatable in the routing collection
153     def get_table(self, exchange):
154         """Get table of bindings for ``exchange``."""
155         localRoutes = frozenset(self.state.exchanges[exchange]['table'])
156         brokerRoutes = self.client.messages.routing.find(
157             {'exchange': exchange}
158         )
159 
160         return localRoutes | frozenset((r['routing_key'],
161                                         r['pattern'],
162                                         r['queue']) for r in brokerRoutes)
163 
164     def _put_fanout(self, exchange, message, **kwargs):
165         """Deliver fanout message."""
166         self.client.messages.broadcast.insert({'payload': dumps(message),
167                                                'queue': exchange})
168 
169     def _queue_bind(self, exchange, routing_key, pattern, queue):
170         if self.typeof(exchange).type == 'fanout':
171             cursor = self.bcast.find(query={'queue': exchange},
172                                      sort=[('$natural', 1)], tailable=True)
173             # Fast forward the cursor past old events
174             self._queue_cursors[queue] = cursor.skip(cursor.count())
175             self._queue_readcounts[queue] = cursor.count()
176             self._fanout_queues[queue] = exchange
177 
178         meta = {'exchange': exchange,
179                 'queue': queue,
180                 'routing_key': routing_key,
181                 'pattern': pattern}
182         self.client.messages.routing.update(meta, meta, upsert=True)
183 
184     def queue_delete(self, queue, **kwargs):
185         self.routing.remove({'queue': queue})
186         super(Channel, self).queue_delete(queue, **kwargs)
187         if queue in self._fanout_queues:
188             self._queue_cursors[queue].close()
189             self._queue_cursors.pop(queue, None)
190             self._fanout_queues.pop(queue, None)
191 
192     @property
193     def client(self):
194         if self._client is None:
195             self._client = self._open()
196         return self._client
197 
198 
199 class Transport(virtual.Transport):
200     Channel = Channel
201 
202     can_parse_url = True
203     polling_interval = 1
204     default_port = DEFAULT_PORT
205     connection_errors = (
206         virtual.Transport.connection_errors + (errors.ConnectionFailure, )
207     )
208     channel_errors = (
209         virtual.Transport.channel_errors + (
210             errors.ConnectionFailure,
211             errors.OperationFailure)
212     )
213     driver_type = 'mongodb'
214     driver_name = 'pymongo'
215 
216     def driver_version(self):
217         return pymongo.version
218
```
--- 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/mongodb.py b/kombu/transport/mongodb.py
--- a/kombu/transport/mongodb.py
+++ b/kombu/transport/mongodb.py
@@ -14,7 +14,7 @@
 
 from pymongo import errors
 from anyjson import loads, dumps
-from pymongo import MongoClient
+from pymongo import MongoClient, uri_parser
 
 from kombu.five import Empty
 from kombu.syn import _detect_environment
@@ -91,38 +91,52 @@
             self.client.messages.remove({'queue': queue})
         return size
 
-    def _open(self, scheme='mongodb://'):
+    def _parse_uri(self, scheme='mongodb://'):
         # See mongodb uri documentation:
-        # http://www.mongodb.org/display/DOCS/Connections
+        # http://docs.mongodb.org/manual/reference/connection-string/
         client = self.connection.client
-        options = client.transport_options
-        hostname = client.hostname or DEFAULT_HOST
-        dbname = client.virtual_host
+        hostname = client.hostname
 
-        if dbname in ['/', None]:
-            dbname = "kombu_default"
         if not hostname.startswith(scheme):
             hostname = scheme + hostname
 
         if not hostname[len(scheme):]:
-            hostname += 'localhost'
-
-        # XXX What does this do?  [ask]
-        urest = hostname[len(scheme):]
-        if '/' in urest:
-            if not client.userid:
-                urest = urest.replace('/' + client.virtual_host, '/')
-                hostname = ''.join([scheme, urest])
-
-        # At this point we expect the hostname to be something like
-        # (considering replica set form too):
-        #
-        #   mongodb://[username:password@]host1[:port1][,host2[:port2],
-        #   ...[,hostN[:portN]]][/[?options]]
-        options.setdefault('auto_start_request', True)
+            hostname += DEFAULT_HOST
+
+        if client.userid and '@' not in hostname:
+            head, tail = hostname.split('://')
+
+            credentials = client.userid
+            if client.password:
+                credentials += ':' + client.password
+
+            hostname = head + '://' + credentials + '@' + tail
+
+        port = client.port if client.port is not None else DEFAULT_PORT
+
+        parsed = uri_parser.parse_uri(hostname, port)
+
+        dbname = parsed['database'] or client.virtual_host
+
+        if dbname in ('/', None):
+            dbname = 'kombu_default'
+
+        options = {'auto_start_request': True,
+                   'ssl': client.ssl,
+                   'connectTimeoutMS': int(client.connect_timeout * 1000)
+                                            if client.connect_timeout else None}
+        options.update(client.transport_options)
+        options.update(parsed['options'])
+
+        return hostname, dbname, options
+
+    def _open(self, scheme='mongodb://'):
+        hostname, dbname, options = self._parse_uri(scheme=scheme)
+
         mongoconn = MongoClient(
-            host=hostname, ssl=client.ssl,
+            host=hostname, ssl=options['ssl'],
             auto_start_request=options['auto_start_request'],
+            connectTimeoutMS=options['connectTimeoutMS'],
             use_greenlets=_detect_environment() != 'default',
         )
         database = getattr(mongoconn, dbname)

{"golden_diff": "diff --git a/kombu/transport/mongodb.py b/kombu/transport/mongodb.py\n--- a/kombu/transport/mongodb.py\n+++ b/kombu/transport/mongodb.py\n@@ -14,7 +14,7 @@\n \n from pymongo import errors\n from anyjson import loads, dumps\n-from pymongo import MongoClient\n+from pymongo import MongoClient, uri_parser\n \n from kombu.five import Empty\n from kombu.syn import _detect_environment\n@@ -91,38 +91,52 @@\n             self.client.messages.remove({'queue': queue})\n         return size\n \n-    def _open(self, scheme='mongodb://'):\n+    def _parse_uri(self, scheme='mongodb://'):\n         # See mongodb uri documentation:\n-        # http://www.mongodb.org/display/DOCS/Connections\n+        # http://docs.mongodb.org/manual/reference/connection-string/\n         client = self.connection.client\n-        options = client.transport_options\n-        hostname = client.hostname or DEFAULT_HOST\n-        dbname = client.virtual_host\n+        hostname = client.hostname\n \n-        if dbname in ['/', None]:\n-            dbname = \"kombu_default\"\n         if not hostname.startswith(scheme):\n             hostname = scheme + hostname\n \n         if not hostname[len(scheme):]:\n-            hostname += 'localhost'\n-\n-        # XXX What does this do?  [ask]\n-        urest = hostname[len(scheme):]\n-        if '/' in urest:\n-            if not client.userid:\n-                urest = urest.replace('/' + client.virtual_host, '/')\n-                hostname = ''.join([scheme, urest])\n-\n-        # At this point we expect the hostname to be something like\n-        # (considering replica set form too):\n-        #\n-        #   mongodb://[username:password@]host1[:port1][,host2[:port2],\n-        #   ...[,hostN[:portN]]][/[?options]]\n-        options.setdefault('auto_start_request', True)\n+            hostname += DEFAULT_HOST\n+\n+        if client.userid and '@' not in hostname:\n+            head, tail = hostname.split('://')\n+\n+            credentials = client.userid\n+            if client.password:\n+                credentials += ':' + client.password\n+\n+            hostname = head + '://' + credentials + '@' + tail\n+\n+        port = client.port if client.port is not None else DEFAULT_PORT\n+\n+        parsed = uri_parser.parse_uri(hostname, port)\n+\n+        dbname = parsed['database'] or client.virtual_host\n+\n+        if dbname in ('/', None):\n+            dbname = 'kombu_default'\n+\n+        options = {'auto_start_request': True,\n+                   'ssl': client.ssl,\n+                   'connectTimeoutMS': int(client.connect_timeout * 1000)\n+                                            if client.connect_timeout else None}\n+        options.update(client.transport_options)\n+        options.update(parsed['options'])\n+\n+        return hostname, dbname, options\n+\n+    def _open(self, scheme='mongodb://'):\n+        hostname, dbname, options = self._parse_uri(scheme=scheme)\n+\n         mongoconn = MongoClient(\n-            host=hostname, ssl=client.ssl,\n+            host=hostname, ssl=options['ssl'],\n             auto_start_request=options['auto_start_request'],\n+            connectTimeoutMS=options['connectTimeoutMS'],\n             use_greenlets=_detect_environment() != 'default',\n         )\n         database = getattr(mongoconn, dbname)\n", "issue": "Mongodb url is broken in Kombu 3.0.1\nUpgrading to version 3.0.1, a mongodb url like mongodb://localhost/wq  fails as the // gets converted to / by some code in transport/mongodb.py  - basically it seems to boil down to this code:-\n\n```\n    if '/' in hostname[10:]:\n        if not client.userid:\n            hostname = hostname.replace('/' + client.virtual_host, '/')\n        else:\n            hostname = hostname.replace('/' + client.virtual_host, '/' + authdb)\n```\n\nby default virtual_host == '/' so basically hostname becomes:-\n\nmongodb:/localhost/wq  instead of mongodb://localhost/wq causing the MongoClient to fail wit a port not an integer error.\n\nright now can work around this by setting the url to mongodb:////localhost/wq but its a kludge.\n\nmongodb transport no longer honors configured port.\nThe mongodb transport in kombu version 2.1.3 (version according to pypi) has changed the BrokerConnection arguments that it uses. We are passing in hostname, port, and transport in as separate BrokerConnection arguments, but the mongodb transport now expects to get all that information out of hostname, which it assumes is a mongodb:// URI.\n\nFor some history, version 1.5.0 changed the parse_url() function so that the port was no longer parsed out of mongodb:// URIs. So we were forced to pass the hostname, port, and transport in as separate BrokerConnection arguments. Now, we are being forced back to passing a mongodb:// URI in as the hostname.\n\n", "before_files": [{"content": "\"\"\"\nkombu.transport.mongodb\n=======================\n\nMongoDB transport.\n\n:copyright: (c) 2010 - 2013 by Flavio Percoco Premoli.\n:license: BSD, see LICENSE for more details.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nimport pymongo\n\nfrom pymongo import errors\nfrom anyjson import loads, dumps\nfrom pymongo import MongoClient\n\nfrom kombu.five import Empty\nfrom kombu.syn import _detect_environment\nfrom kombu.utils.encoding import bytes_to_str\n\nfrom . import virtual\n\nDEFAULT_HOST = '127.0.0.1'\nDEFAULT_PORT = 27017\n\n__author__ = \"\"\"\\\nFlavio [FlaPer87] Percoco Premoli <flaper87@flaper87.org>;\\\nScott Lyons <scottalyons@gmail.com>;\\\n\"\"\"\n\n\nclass Channel(virtual.Channel):\n    _client = None\n    supports_fanout = True\n    _fanout_queues = {}\n\n    def __init__(self, *vargs, **kwargs):\n        super_ = super(Channel, self)\n        super_.__init__(*vargs, **kwargs)\n\n        self._queue_cursors = {}\n        self._queue_readcounts = {}\n\n    def _new_queue(self, queue, **kwargs):\n        pass\n\n    def _get(self, queue):\n        try:\n            if queue in self._fanout_queues:\n                msg = next(self._queue_cursors[queue])\n                self._queue_readcounts[queue] += 1\n                return loads(bytes_to_str(msg['payload']))\n            else:\n                msg = self.client.command(\n                    'findandmodify', 'messages',\n                    query={'queue': queue},\n                    sort={'_id': pymongo.ASCENDING}, remove=True,\n                )\n        except errors.OperationFailure as exc:\n            if 'No matching object found' in exc.args[0]:\n                raise Empty()\n            raise\n        except StopIteration:\n            raise Empty()\n\n        # as of mongo 2.0 empty results won't raise an error\n        if msg['value'] is None:\n            raise Empty()\n        return loads(bytes_to_str(msg['value']['payload']))\n\n    def _size(self, queue):\n        if queue in self._fanout_queues:\n            return (self._queue_cursors[queue].count() -\n                    self._queue_readcounts[queue])\n\n        return self.client.messages.find({'queue': queue}).count()\n\n    def _put(self, queue, message, **kwargs):\n        self.client.messages.insert({'payload': dumps(message),\n                                     'queue': queue})\n\n    def _purge(self, queue):\n        size = self._size(queue)\n        if queue in self._fanout_queues:\n            cursor = self._queue_cursors[queue]\n            cursor.rewind()\n            self._queue_cursors[queue] = cursor.skip(cursor.count())\n        else:\n            self.client.messages.remove({'queue': queue})\n        return size\n\n    def _open(self, scheme='mongodb://'):\n        # See mongodb uri documentation:\n        # http://www.mongodb.org/display/DOCS/Connections\n        client = self.connection.client\n        options = client.transport_options\n        hostname = client.hostname or DEFAULT_HOST\n        dbname = client.virtual_host\n\n        if dbname in ['/', None]:\n            dbname = \"kombu_default\"\n        if not hostname.startswith(scheme):\n            hostname = scheme + hostname\n\n        if not hostname[len(scheme):]:\n            hostname += 'localhost'\n\n        # XXX What does this do?  [ask]\n        urest = hostname[len(scheme):]\n        if '/' in urest:\n            if not client.userid:\n                urest = urest.replace('/' + client.virtual_host, '/')\n                hostname = ''.join([scheme, urest])\n\n        # At this point we expect the hostname to be something like\n        # (considering replica set form too):\n        #\n        #   mongodb://[username:password@]host1[:port1][,host2[:port2],\n        #   ...[,hostN[:portN]]][/[?options]]\n        options.setdefault('auto_start_request', True)\n        mongoconn = MongoClient(\n            host=hostname, ssl=client.ssl,\n            auto_start_request=options['auto_start_request'],\n            use_greenlets=_detect_environment() != 'default',\n        )\n        database = getattr(mongoconn, dbname)\n\n        version = mongoconn.server_info()['version']\n        if tuple(map(int, version.split('.')[:2])) < (1, 3):\n            raise NotImplementedError(\n                'Kombu requires MongoDB version 1.3+ (server is {0})'.format(\n                    version))\n\n        self.db = database\n        col = database.messages\n        col.ensure_index([('queue', 1), ('_id', 1)], background=True)\n\n        if 'messages.broadcast' not in database.collection_names():\n            capsize = options.get('capped_queue_size') or 100000\n            database.create_collection('messages.broadcast',\n                                       size=capsize, capped=True)\n\n        self.bcast = getattr(database, 'messages.broadcast')\n        self.bcast.ensure_index([('queue', 1)])\n\n        self.routing = getattr(database, 'messages.routing')\n        self.routing.ensure_index([('queue', 1), ('exchange', 1)])\n        return database\n\n    #TODO: Store a more complete exchange metatable in the routing collection\n    def get_table(self, exchange):\n        \"\"\"Get table of bindings for ``exchange``.\"\"\"\n        localRoutes = frozenset(self.state.exchanges[exchange]['table'])\n        brokerRoutes = self.client.messages.routing.find(\n            {'exchange': exchange}\n        )\n\n        return localRoutes | frozenset((r['routing_key'],\n                                        r['pattern'],\n                                        r['queue']) for r in brokerRoutes)\n\n    def _put_fanout(self, exchange, message, **kwargs):\n        \"\"\"Deliver fanout message.\"\"\"\n        self.client.messages.broadcast.insert({'payload': dumps(message),\n                                               'queue': exchange})\n\n    def _queue_bind(self, exchange, routing_key, pattern, queue):\n        if self.typeof(exchange).type == 'fanout':\n            cursor = self.bcast.find(query={'queue': exchange},\n                                     sort=[('$natural', 1)], tailable=True)\n            # Fast forward the cursor past old events\n            self._queue_cursors[queue] = cursor.skip(cursor.count())\n            self._queue_readcounts[queue] = cursor.count()\n            self._fanout_queues[queue] = exchange\n\n        meta = {'exchange': exchange,\n                'queue': queue,\n                'routing_key': routing_key,\n                'pattern': pattern}\n        self.client.messages.routing.update(meta, meta, upsert=True)\n\n    def queue_delete(self, queue, **kwargs):\n        self.routing.remove({'queue': queue})\n        super(Channel, self).queue_delete(queue, **kwargs)\n        if queue in self._fanout_queues:\n            self._queue_cursors[queue].close()\n            self._queue_cursors.pop(queue, None)\n            self._fanout_queues.pop(queue, None)\n\n    @property\n    def client(self):\n        if self._client is None:\n            self._client = self._open()\n        return self._client\n\n\nclass Transport(virtual.Transport):\n    Channel = Channel\n\n    can_parse_url = True\n    polling_interval = 1\n    default_port = DEFAULT_PORT\n    connection_errors = (\n        virtual.Transport.connection_errors + (errors.ConnectionFailure, )\n    )\n    channel_errors = (\n        virtual.Transport.channel_errors + (\n            errors.ConnectionFailure,\n            errors.OperationFailure)\n    )\n    driver_type = 'mongodb'\n    driver_name = 'pymongo'\n\n    def driver_version(self):\n        return pymongo.version\n", "path": "kombu/transport/mongodb.py"}], "after_files": [{"content": "\"\"\"\nkombu.transport.mongodb\n=======================\n\nMongoDB transport.\n\n:copyright: (c) 2010 - 2013 by Flavio Percoco Premoli.\n:license: BSD, see LICENSE for more details.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nimport pymongo\n\nfrom pymongo import errors\nfrom anyjson import loads, dumps\nfrom pymongo import MongoClient, uri_parser\n\nfrom kombu.five import Empty\nfrom kombu.syn import _detect_environment\nfrom kombu.utils.encoding import bytes_to_str\n\nfrom . import virtual\n\nDEFAULT_HOST = '127.0.0.1'\nDEFAULT_PORT = 27017\n\n__author__ = \"\"\"\\\nFlavio [FlaPer87] Percoco Premoli <flaper87@flaper87.org>;\\\nScott Lyons <scottalyons@gmail.com>;\\\n\"\"\"\n\n\nclass Channel(virtual.Channel):\n    _client = None\n    supports_fanout = True\n    _fanout_queues = {}\n\n    def __init__(self, *vargs, **kwargs):\n        super_ = super(Channel, self)\n        super_.__init__(*vargs, **kwargs)\n\n        self._queue_cursors = {}\n        self._queue_readcounts = {}\n\n    def _new_queue(self, queue, **kwargs):\n        pass\n\n    def _get(self, queue):\n        try:\n            if queue in self._fanout_queues:\n                msg = next(self._queue_cursors[queue])\n                self._queue_readcounts[queue] += 1\n                return loads(bytes_to_str(msg['payload']))\n            else:\n                msg = self.client.command(\n                    'findandmodify', 'messages',\n                    query={'queue': queue},\n                    sort={'_id': pymongo.ASCENDING}, remove=True,\n                )\n        except errors.OperationFailure as exc:\n            if 'No matching object found' in exc.args[0]:\n                raise Empty()\n            raise\n        except StopIteration:\n            raise Empty()\n\n        # as of mongo 2.0 empty results won't raise an error\n        if msg['value'] is None:\n            raise Empty()\n        return loads(bytes_to_str(msg['value']['payload']))\n\n    def _size(self, queue):\n        if queue in self._fanout_queues:\n            return (self._queue_cursors[queue].count() -\n                    self._queue_readcounts[queue])\n\n        return self.client.messages.find({'queue': queue}).count()\n\n    def _put(self, queue, message, **kwargs):\n        self.client.messages.insert({'payload': dumps(message),\n                                     'queue': queue})\n\n    def _purge(self, queue):\n        size = self._size(queue)\n        if queue in self._fanout_queues:\n            cursor = self._queue_cursors[queue]\n            cursor.rewind()\n            self._queue_cursors[queue] = cursor.skip(cursor.count())\n        else:\n            self.client.messages.remove({'queue': queue})\n        return size\n\n    def _parse_uri(self, scheme='mongodb://'):\n        # See mongodb uri documentation:\n        # http://docs.mongodb.org/manual/reference/connection-string/\n        client = self.connection.client\n        hostname = client.hostname\n\n        if not hostname.startswith(scheme):\n            hostname = scheme + hostname\n\n        if not hostname[len(scheme):]:\n            hostname += DEFAULT_HOST\n\n        if client.userid and '@' not in hostname:\n            head, tail = hostname.split('://')\n\n            credentials = client.userid\n            if client.password:\n                credentials += ':' + client.password\n\n            hostname = head + '://' + credentials + '@' + tail\n\n        port = client.port if client.port is not None else DEFAULT_PORT\n\n        parsed = uri_parser.parse_uri(hostname, port)\n\n        dbname = parsed['database'] or client.virtual_host\n\n        if dbname in ('/', None):\n            dbname = 'kombu_default'\n\n        options = {'auto_start_request': True,\n                   'ssl': client.ssl,\n                   'connectTimeoutMS': int(client.connect_timeout * 1000)\n                                            if client.connect_timeout else None}\n        options.update(client.transport_options)\n        options.update(parsed['options'])\n\n        return hostname, dbname, options\n\n    def _open(self, scheme='mongodb://'):\n        hostname, dbname, options = self._parse_uri(scheme=scheme)\n\n        mongoconn = MongoClient(\n            host=hostname, ssl=options['ssl'],\n            auto_start_request=options['auto_start_request'],\n            connectTimeoutMS=options['connectTimeoutMS'],\n            use_greenlets=_detect_environment() != 'default',\n        )\n        database = getattr(mongoconn, dbname)\n\n        version = mongoconn.server_info()['version']\n        if tuple(map(int, version.split('.')[:2])) < (1, 3):\n            raise NotImplementedError(\n                'Kombu requires MongoDB version 1.3+ (server is {0})'.format(\n                    version))\n\n        self.db = database\n        col = database.messages\n        col.ensure_index([('queue', 1), ('_id', 1)], background=True)\n\n        if 'messages.broadcast' not in database.collection_names():\n            capsize = options.get('capped_queue_size') or 100000\n            database.create_collection('messages.broadcast',\n                                       size=capsize, capped=True)\n\n        self.bcast = getattr(database, 'messages.broadcast')\n        self.bcast.ensure_index([('queue', 1)])\n\n        self.routing = getattr(database, 'messages.routing')\n        self.routing.ensure_index([('queue', 1), ('exchange', 1)])\n        return database\n\n    #TODO: Store a more complete exchange metatable in the routing collection\n    def get_table(self, exchange):\n        \"\"\"Get table of bindings for ``exchange``.\"\"\"\n        localRoutes = frozenset(self.state.exchanges[exchange]['table'])\n        brokerRoutes = self.client.messages.routing.find(\n            {'exchange': exchange}\n        )\n\n        return localRoutes | frozenset((r['routing_key'],\n                                        r['pattern'],\n                                        r['queue']) for r in brokerRoutes)\n\n    def _put_fanout(self, exchange, message, **kwargs):\n        \"\"\"Deliver fanout message.\"\"\"\n        self.client.messages.broadcast.insert({'payload': dumps(message),\n                                               'queue': exchange})\n\n    def _queue_bind(self, exchange, routing_key, pattern, queue):\n        if self.typeof(exchange).type == 'fanout':\n            cursor = self.bcast.find(query={'queue': exchange},\n                                     sort=[('$natural', 1)], tailable=True)\n            # Fast forward the cursor past old events\n            self._queue_cursors[queue] = cursor.skip(cursor.count())\n            self._queue_readcounts[queue] = cursor.count()\n            self._fanout_queues[queue] = exchange\n\n        meta = {'exchange': exchange,\n                'queue': queue,\n                'routing_key': routing_key,\n                'pattern': pattern}\n        self.client.messages.routing.update(meta, meta, upsert=True)\n\n    def queue_delete(self, queue, **kwargs):\n        self.routing.remove({'queue': queue})\n        super(Channel, self).queue_delete(queue, **kwargs)\n        if queue in self._fanout_queues:\n            self._queue_cursors[queue].close()\n            self._queue_cursors.pop(queue, None)\n            self._fanout_queues.pop(queue, None)\n\n    @property\n    def client(self):\n        if self._client is None:\n            self._client = self._open()\n        return self._client\n\n\nclass Transport(virtual.Transport):\n    Channel = Channel\n\n    can_parse_url = True\n    polling_interval = 1\n    default_port = DEFAULT_PORT\n    connection_errors = (\n        virtual.Transport.connection_errors + (errors.ConnectionFailure, )\n    )\n    channel_errors = (\n        virtual.Transport.channel_errors + (\n            errors.ConnectionFailure,\n            errors.OperationFailure)\n    )\n    driver_type = 'mongodb'\n    driver_name = 'pymongo'\n\n    def driver_version(self):\n        return pymongo.version\n", "path": "kombu/transport/mongodb.py"}]}