Fix MongoDB transport connection string and options handling

1 files changed, 28 insertions, 22 deletions

diff --git a/kombu/transport/mongodb.py b/kombu/transport/mongodb.py
index b716f978..9e0563f2 100644
--- a/kombu/transport/mongodb.py
+++ b/kombu/transport/mongodb.py
@@ -14,7 +14,7 @@ import pymongo
 
 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
@@ -93,14 +93,9 @@ class Channel(virtual.Channel):
 
     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
-
-        if dbname in ['/', None]:
-            dbname = "kombu_default"
+        hostname = client.hostname
 
         if not hostname.startswith(scheme):
             hostname = scheme + hostname
@@ -108,20 +103,30 @@ class Channel(virtual.Channel):
         if not hostname[len(scheme):]:
             hostname += DEFAULT_HOST
 
-        # 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)
-        options.setdefault('ssl', client.ssl)
+        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
 
@@ -131,6 +136,7 @@ class Channel(virtual.Channel):
         mongoconn = MongoClient(
             host=hostname, ssl=options['ssl'],
             auto_start_request=options['auto_start_request'],
+            connectTimeoutMS=options['connectTimeoutMS'],
             use_greenlets=_detect_environment() != 'default',
         )
         database = getattr(mongoconn, dbname)