We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
unexpected keyword argument 'query' when using pymongo 3.0
```
  File "/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/kombu/transport/virtual/__init__.py", line 487, in queue_bind
    self._queue_bind(exchange, *meta)
  File "/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/kombu/transport/mongodb.py", line 244, in _queue_bind
    self.create_broadcast_cursor(exchange, routing_key, pattern, queue)
  File "/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/kombu/transport/mongodb.py", line 302, in create_broadcast_cursor
    tailable=True,
  File "/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/pymongo/collection.py", line 929, in find
    return Cursor(self, *args, **kwargs)
TypeError: __init__() got an unexpected keyword argument 'query'
```

https://github.com/celery/kombu/blob/8e6aed9fcf978b6c34108c0d37b720bd125f0352/kombu/transport/mongodb.py#L297-L300

query and tailable are invalid arguments with pymongo 3.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/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 pymongo import MongoClient, uri_parser
17 
18 from kombu.five import Empty
19 from kombu.syn import _detect_environment
20 from kombu.utils.encoding import bytes_to_str
21 from kombu.utils.json import loads, dumps
22 
23 from . import virtual
24 
25 DEFAULT_HOST = '127.0.0.1'
26 DEFAULT_PORT = 27017
27 
28 DEFAULT_MESSAGES_COLLECTION = 'messages'
29 DEFAULT_ROUTING_COLLECTION = 'messages.routing'
30 DEFAULT_BROADCAST_COLLECTION = 'messages.broadcast'
31 
32 
33 class BroadcastCursor(object):
34     """Cursor for broadcast queues."""
35 
36     def __init__(self, cursor):
37         self._cursor = cursor
38 
39         self.purge(rewind=False)
40 
41     def get_size(self):
42         return self._cursor.count() - self._offset
43 
44     def close(self):
45         self._cursor.close()
46 
47     def purge(self, rewind=True):
48         if rewind:
49             self._cursor.rewind()
50 
51         # Fast forward the cursor past old events
52         self._offset = self._cursor.count()
53         self._cursor = self._cursor.skip(self._offset)
54 
55     def __iter__(self):
56         return self
57 
58     def __next__(self):
59         while True:
60             try:
61                 msg = next(self._cursor)
62             except pymongo.errors.OperationFailure as exc:
63                 # In some cases tailed cursor can become invalid
64                 # and have to be reinitalized
65                 if 'not valid at server' in exc.message:
66                     self.purge()
67 
68                     continue
69 
70                 raise
71             else:
72                 break
73 
74         self._offset += 1
75 
76         return msg
77     next = __next__
78 
79 
80 class Channel(virtual.Channel):
81     _client = None
82     supports_fanout = True
83     _fanout_queues = {}
84 
85     def __init__(self, *vargs, **kwargs):
86         super(Channel, self).__init__(*vargs, **kwargs)
87 
88         self._broadcast_cursors = {}
89 
90         # Evaluate connection
91         self._create_client()
92 
93     def _new_queue(self, queue, **kwargs):
94         pass
95 
96     def _get(self, queue):
97         if queue in self._fanout_queues:
98             try:
99                 msg = next(self.get_broadcast_cursor(queue))
100             except StopIteration:
101                 msg = None
102         else:
103             msg = self.get_messages().find_and_modify(
104                 query={'queue': queue},
105                 sort=[('priority', pymongo.ASCENDING),
106                       ('_id', pymongo.ASCENDING)],
107                 remove=True,
108             )
109 
110         if msg is None:
111             raise Empty()
112 
113         return loads(bytes_to_str(msg['payload']))
114 
115     def _size(self, queue):
116         if queue in self._fanout_queues:
117             return self.get_broadcast_cursor(queue).get_size()
118 
119         return self.get_messages().find({'queue': queue}).count()
120 
121     def _put(self, queue, message, **kwargs):
122         self.get_messages().insert({
123             'payload': dumps(message),
124             'queue': queue,
125             'priority': self._get_message_priority(message, reverse=True),
126         })
127 
128     def _purge(self, queue):
129         size = self._size(queue)
130 
131         if queue in self._fanout_queues:
132             self.get_broadcaset_cursor(queue).purge()
133         else:
134             self.get_messages().remove({'queue': queue})
135 
136         return size
137 
138     def _parse_uri(self, scheme='mongodb://'):
139         # See mongodb uri documentation:
140         # http://docs.mongodb.org/manual/reference/connection-string/
141         client = self.connection.client
142         hostname = client.hostname
143 
144         if not hostname.startswith(scheme):
145             hostname = scheme + hostname
146 
147         if not hostname[len(scheme):]:
148             hostname += DEFAULT_HOST
149 
150         if client.userid and '@' not in hostname:
151             head, tail = hostname.split('://')
152 
153             credentials = client.userid
154             if client.password:
155                 credentials += ':' + client.password
156 
157             hostname = head + '://' + credentials + '@' + tail
158 
159         port = client.port if client.port is not None else DEFAULT_PORT
160 
161         parsed = uri_parser.parse_uri(hostname, port)
162 
163         dbname = parsed['database'] or client.virtual_host
164 
165         if dbname in ('/', None):
166             dbname = 'kombu_default'
167 
168         options = {
169             'auto_start_request': True,
170             'ssl': client.ssl,
171             'connectTimeoutMS': (int(client.connect_timeout * 1000)
172                                  if client.connect_timeout else None),
173         }
174         options.update(client.transport_options)
175         options.update(parsed['options'])
176 
177         return hostname, dbname, options
178 
179     def _prepare_client_options(self, options):
180         if pymongo.version_tuple >= (3, ):
181             options.pop('auto_start_request', None)
182 
183     def _open(self, scheme='mongodb://'):
184         hostname, dbname, options = self._parse_uri(scheme=scheme)
185 
186         self._prepare_client_options(options)
187         mongoconn = MongoClient(
188             host=hostname, ssl=options['ssl'],
189             auto_start_request=options['auto_start_request'],
190             connectTimeoutMS=options['connectTimeoutMS'],
191             use_greenlets=_detect_environment() != 'default',
192         )
193         database = mongoconn[dbname]
194 
195         version = mongoconn.server_info()['version']
196         if tuple(map(int, version.split('.')[:2])) < (1, 3):
197             raise NotImplementedError(
198                 'Kombu requires MongoDB version 1.3+ (server is {0})'.format(
199                     version))
200 
201         self._create_broadcast(database, options)
202 
203         self._client = database
204 
205     def _create_broadcast(self, database, options):
206         '''Create capped collection for broadcast messages.'''
207         if DEFAULT_BROADCAST_COLLECTION in database.collection_names():
208             return
209 
210         capsize = options.get('capped_queue_size') or 100000
211         database.create_collection(DEFAULT_BROADCAST_COLLECTION,
212                                    size=capsize, capped=True)
213 
214     def _ensure_indexes(self):
215         '''Ensure indexes on collections.'''
216         self.get_messages().ensure_index(
217             [('queue', 1), ('priority', 1), ('_id', 1)], background=True,
218         )
219         self.get_broadcast().ensure_index([('queue', 1)])
220         self.get_routing().ensure_index([('queue', 1), ('exchange', 1)])
221 
222     def get_table(self, exchange):
223         """Get table of bindings for ``exchange``."""
224         # TODO Store a more complete exchange metatable in the
225         #      routing collection
226         localRoutes = frozenset(self.state.exchanges[exchange]['table'])
227         brokerRoutes = self.get_messages().routing.find(
228             {'exchange': exchange}
229         )
230 
231         return localRoutes | frozenset((r['routing_key'],
232                                         r['pattern'],
233                                         r['queue']) for r in brokerRoutes)
234 
235     def _put_fanout(self, exchange, message, routing_key, **kwargs):
236         """Deliver fanout message."""
237         self.get_broadcast().insert({'payload': dumps(message),
238                                      'queue': exchange})
239 
240     def _queue_bind(self, exchange, routing_key, pattern, queue):
241         if self.typeof(exchange).type == 'fanout':
242             self.create_broadcast_cursor(exchange, routing_key, pattern, queue)
243             self._fanout_queues[queue] = exchange
244 
245         meta = {'exchange': exchange,
246                 'queue': queue,
247                 'routing_key': routing_key,
248                 'pattern': pattern}
249         self.get_routing().update(meta, meta, upsert=True)
250 
251     def queue_delete(self, queue, **kwargs):
252         self.get_routing().remove({'queue': queue})
253 
254         super(Channel, self).queue_delete(queue, **kwargs)
255 
256         if queue in self._fanout_queues:
257             try:
258                 cursor = self._broadcast_cursors.pop(queue)
259             except KeyError:
260                 pass
261             else:
262                 cursor.close()
263 
264                 self._fanout_queues.pop(queue)
265 
266     def _create_client(self):
267         self._open()
268         self._ensure_indexes()
269 
270     @property
271     def client(self):
272         if self._client is None:
273             self._create_client()
274         return self._client
275 
276     def get_messages(self):
277         return self.client[DEFAULT_MESSAGES_COLLECTION]
278 
279     def get_routing(self):
280         return self.client[DEFAULT_ROUTING_COLLECTION]
281 
282     def get_broadcast(self):
283         return self.client[DEFAULT_BROADCAST_COLLECTION]
284 
285     def get_broadcast_cursor(self, queue):
286         try:
287             return self._broadcast_cursors[queue]
288         except KeyError:
289             # Cursor may be absent when Channel created more than once.
290             # _fanout_queues is a class-level mutable attribute so it's
291             # shared over all Channel instances.
292             return self.create_broadcast_cursor(
293                 self._fanout_queues[queue], None, None, queue,
294             )
295 
296     def create_broadcast_cursor(self, exchange, routing_key, pattern, queue):
297         cursor = self.get_broadcast().find(
298             query={'queue': exchange},
299             sort=[('$natural', 1)],
300             tailable=True,
301         )
302         ret = self._broadcast_cursors[queue] = BroadcastCursor(cursor)
303         return ret
304 
305 
306 class Transport(virtual.Transport):
307     Channel = Channel
308 
309     can_parse_url = True
310     polling_interval = 1
311     default_port = DEFAULT_PORT
312     connection_errors = (
313         virtual.Transport.connection_errors + (errors.ConnectionFailure, )
314     )
315     channel_errors = (
316         virtual.Transport.channel_errors + (
317             errors.ConnectionFailure,
318             errors.OperationFailure)
319     )
320     driver_type = 'mongodb'
321     driver_name = 'pymongo'
322 
323     implements = virtual.Transport.implements.extend(
324         exchange_types=frozenset(['direct', 'topic', 'fanout']),
325     )
326 
327     def driver_version(self):
328         return pymongo.version
329
```
--- 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,6 +14,7 @@
 
 from pymongo import errors
 from pymongo import MongoClient, uri_parser
+from pymongo.cursor import CursorType
 
 from kombu.five import Empty
 from kombu.syn import _detect_environment
@@ -179,17 +180,23 @@
     def _prepare_client_options(self, options):
         if pymongo.version_tuple >= (3, ):
             options.pop('auto_start_request', None)
+        return options
 
     def _open(self, scheme='mongodb://'):
         hostname, dbname, options = self._parse_uri(scheme=scheme)
 
-        self._prepare_client_options(options)
-        mongoconn = MongoClient(
-            host=hostname, ssl=options['ssl'],
-            auto_start_request=options['auto_start_request'],
-            connectTimeoutMS=options['connectTimeoutMS'],
-            use_greenlets=_detect_environment() != 'default',
-        )
+        conf = self._prepare_client_options(options)
+        conf['host'] = hostname
+
+        env = _detect_environment()
+        if env == 'gevent':
+            from gevent import monkey
+            monkey.patch_all()
+        elif env == 'eventlet':
+            from eventlet import monkey_patch
+            monkey_patch()
+
+        mongoconn = MongoClient(**conf)
         database = mongoconn[dbname]
 
         version = mongoconn.server_info()['version']
@@ -294,11 +301,18 @@
             )
 
     def create_broadcast_cursor(self, exchange, routing_key, pattern, queue):
-        cursor = self.get_broadcast().find(
-            query={'queue': exchange},
-            sort=[('$natural', 1)],
-            tailable=True,
-        )
+        if pymongo.version_tuple >= (3, ):
+            query = dict(filter={'queue': exchange},
+                         sort=[('$natural', 1)],
+                         cursor_type=CursorType.TAILABLE
+                         )
+        else:
+            query = dict(query={'queue': exchange},
+                         sort=[('$natural', 1)],
+                         tailable=True
+                         )
+
+        cursor = self.get_broadcast().find(**query)
         ret = self._broadcast_cursors[queue] = BroadcastCursor(cursor)
         return ret

{"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,6 +14,7 @@\n \n from pymongo import errors\n from pymongo import MongoClient, uri_parser\n+from pymongo.cursor import CursorType\n \n from kombu.five import Empty\n from kombu.syn import _detect_environment\n@@ -179,17 +180,23 @@\n     def _prepare_client_options(self, options):\n         if pymongo.version_tuple >= (3, ):\n             options.pop('auto_start_request', None)\n+        return options\n \n     def _open(self, scheme='mongodb://'):\n         hostname, dbname, options = self._parse_uri(scheme=scheme)\n \n-        self._prepare_client_options(options)\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+        conf = self._prepare_client_options(options)\n+        conf['host'] = hostname\n+\n+        env = _detect_environment()\n+        if env == 'gevent':\n+            from gevent import monkey\n+            monkey.patch_all()\n+        elif env == 'eventlet':\n+            from eventlet import monkey_patch\n+            monkey_patch()\n+\n+        mongoconn = MongoClient(**conf)\n         database = mongoconn[dbname]\n \n         version = mongoconn.server_info()['version']\n@@ -294,11 +301,18 @@\n             )\n \n     def create_broadcast_cursor(self, exchange, routing_key, pattern, queue):\n-        cursor = self.get_broadcast().find(\n-            query={'queue': exchange},\n-            sort=[('$natural', 1)],\n-            tailable=True,\n-        )\n+        if pymongo.version_tuple >= (3, ):\n+            query = dict(filter={'queue': exchange},\n+                         sort=[('$natural', 1)],\n+                         cursor_type=CursorType.TAILABLE\n+                         )\n+        else:\n+            query = dict(query={'queue': exchange},\n+                         sort=[('$natural', 1)],\n+                         tailable=True\n+                         )\n+\n+        cursor = self.get_broadcast().find(**query)\n         ret = self._broadcast_cursors[queue] = BroadcastCursor(cursor)\n         return ret\n", "issue": "unexpected keyword argument 'query' when using pymongo 3.0\n```\n  File \"/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/kombu/transport/virtual/__init__.py\", line 487, in queue_bind\n    self._queue_bind(exchange, *meta)\n  File \"/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/kombu/transport/mongodb.py\", line 244, in _queue_bind\n    self.create_broadcast_cursor(exchange, routing_key, pattern, queue)\n  File \"/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/kombu/transport/mongodb.py\", line 302, in create_broadcast_cursor\n    tailable=True,\n  File \"/opt/ampli/apps/marvintaskmgr/venv/lib/python2.7/site-packages/pymongo/collection.py\", line 929, in find\n    return Cursor(self, *args, **kwargs)\nTypeError: __init__() got an unexpected keyword argument 'query'\n```\n\nhttps://github.com/celery/kombu/blob/8e6aed9fcf978b6c34108c0d37b720bd125f0352/kombu/transport/mongodb.py#L297-L300\n\nquery and tailable are invalid arguments with pymongo 3.0\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 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\nfrom kombu.utils.json import loads, dumps\n\nfrom . import virtual\n\nDEFAULT_HOST = '127.0.0.1'\nDEFAULT_PORT = 27017\n\nDEFAULT_MESSAGES_COLLECTION = 'messages'\nDEFAULT_ROUTING_COLLECTION = 'messages.routing'\nDEFAULT_BROADCAST_COLLECTION = 'messages.broadcast'\n\n\nclass BroadcastCursor(object):\n    \"\"\"Cursor for broadcast queues.\"\"\"\n\n    def __init__(self, cursor):\n        self._cursor = cursor\n\n        self.purge(rewind=False)\n\n    def get_size(self):\n        return self._cursor.count() - self._offset\n\n    def close(self):\n        self._cursor.close()\n\n    def purge(self, rewind=True):\n        if rewind:\n            self._cursor.rewind()\n\n        # Fast forward the cursor past old events\n        self._offset = self._cursor.count()\n        self._cursor = self._cursor.skip(self._offset)\n\n    def __iter__(self):\n        return self\n\n    def __next__(self):\n        while True:\n            try:\n                msg = next(self._cursor)\n            except pymongo.errors.OperationFailure as exc:\n                # In some cases tailed cursor can become invalid\n                # and have to be reinitalized\n                if 'not valid at server' in exc.message:\n                    self.purge()\n\n                    continue\n\n                raise\n            else:\n                break\n\n        self._offset += 1\n\n        return msg\n    next = __next__\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(Channel, self).__init__(*vargs, **kwargs)\n\n        self._broadcast_cursors = {}\n\n        # Evaluate connection\n        self._create_client()\n\n    def _new_queue(self, queue, **kwargs):\n        pass\n\n    def _get(self, queue):\n        if queue in self._fanout_queues:\n            try:\n                msg = next(self.get_broadcast_cursor(queue))\n            except StopIteration:\n                msg = None\n        else:\n            msg = self.get_messages().find_and_modify(\n                query={'queue': queue},\n                sort=[('priority', pymongo.ASCENDING),\n                      ('_id', pymongo.ASCENDING)],\n                remove=True,\n            )\n\n        if msg is None:\n            raise Empty()\n\n        return loads(bytes_to_str(msg['payload']))\n\n    def _size(self, queue):\n        if queue in self._fanout_queues:\n            return self.get_broadcast_cursor(queue).get_size()\n\n        return self.get_messages().find({'queue': queue}).count()\n\n    def _put(self, queue, message, **kwargs):\n        self.get_messages().insert({\n            'payload': dumps(message),\n            'queue': queue,\n            'priority': self._get_message_priority(message, reverse=True),\n        })\n\n    def _purge(self, queue):\n        size = self._size(queue)\n\n        if queue in self._fanout_queues:\n            self.get_broadcaset_cursor(queue).purge()\n        else:\n            self.get_messages().remove({'queue': queue})\n\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 = {\n            'auto_start_request': True,\n            'ssl': client.ssl,\n            'connectTimeoutMS': (int(client.connect_timeout * 1000)\n                                 if client.connect_timeout else None),\n        }\n        options.update(client.transport_options)\n        options.update(parsed['options'])\n\n        return hostname, dbname, options\n\n    def _prepare_client_options(self, options):\n        if pymongo.version_tuple >= (3, ):\n            options.pop('auto_start_request', None)\n\n    def _open(self, scheme='mongodb://'):\n        hostname, dbname, options = self._parse_uri(scheme=scheme)\n\n        self._prepare_client_options(options)\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 = 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._create_broadcast(database, options)\n\n        self._client = database\n\n    def _create_broadcast(self, database, options):\n        '''Create capped collection for broadcast messages.'''\n        if DEFAULT_BROADCAST_COLLECTION in database.collection_names():\n            return\n\n        capsize = options.get('capped_queue_size') or 100000\n        database.create_collection(DEFAULT_BROADCAST_COLLECTION,\n                                   size=capsize, capped=True)\n\n    def _ensure_indexes(self):\n        '''Ensure indexes on collections.'''\n        self.get_messages().ensure_index(\n            [('queue', 1), ('priority', 1), ('_id', 1)], background=True,\n        )\n        self.get_broadcast().ensure_index([('queue', 1)])\n        self.get_routing().ensure_index([('queue', 1), ('exchange', 1)])\n\n    def get_table(self, exchange):\n        \"\"\"Get table of bindings for ``exchange``.\"\"\"\n        # TODO Store a more complete exchange metatable in the\n        #      routing collection\n        localRoutes = frozenset(self.state.exchanges[exchange]['table'])\n        brokerRoutes = self.get_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, routing_key, **kwargs):\n        \"\"\"Deliver fanout message.\"\"\"\n        self.get_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            self.create_broadcast_cursor(exchange, routing_key, pattern, queue)\n            self._fanout_queues[queue] = exchange\n\n        meta = {'exchange': exchange,\n                'queue': queue,\n                'routing_key': routing_key,\n                'pattern': pattern}\n        self.get_routing().update(meta, meta, upsert=True)\n\n    def queue_delete(self, queue, **kwargs):\n        self.get_routing().remove({'queue': queue})\n\n        super(Channel, self).queue_delete(queue, **kwargs)\n\n        if queue in self._fanout_queues:\n            try:\n                cursor = self._broadcast_cursors.pop(queue)\n            except KeyError:\n                pass\n            else:\n                cursor.close()\n\n                self._fanout_queues.pop(queue)\n\n    def _create_client(self):\n        self._open()\n        self._ensure_indexes()\n\n    @property\n    def client(self):\n        if self._client is None:\n            self._create_client()\n        return self._client\n\n    def get_messages(self):\n        return self.client[DEFAULT_MESSAGES_COLLECTION]\n\n    def get_routing(self):\n        return self.client[DEFAULT_ROUTING_COLLECTION]\n\n    def get_broadcast(self):\n        return self.client[DEFAULT_BROADCAST_COLLECTION]\n\n    def get_broadcast_cursor(self, queue):\n        try:\n            return self._broadcast_cursors[queue]\n        except KeyError:\n            # Cursor may be absent when Channel created more than once.\n            # _fanout_queues is a class-level mutable attribute so it's\n            # shared over all Channel instances.\n            return self.create_broadcast_cursor(\n                self._fanout_queues[queue], None, None, queue,\n            )\n\n    def create_broadcast_cursor(self, exchange, routing_key, pattern, queue):\n        cursor = self.get_broadcast().find(\n            query={'queue': exchange},\n            sort=[('$natural', 1)],\n            tailable=True,\n        )\n        ret = self._broadcast_cursors[queue] = BroadcastCursor(cursor)\n        return ret\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    implements = virtual.Transport.implements.extend(\n        exchange_types=frozenset(['direct', 'topic', 'fanout']),\n    )\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 pymongo import MongoClient, uri_parser\nfrom pymongo.cursor import CursorType\n\nfrom kombu.five import Empty\nfrom kombu.syn import _detect_environment\nfrom kombu.utils.encoding import bytes_to_str\nfrom kombu.utils.json import loads, dumps\n\nfrom . import virtual\n\nDEFAULT_HOST = '127.0.0.1'\nDEFAULT_PORT = 27017\n\nDEFAULT_MESSAGES_COLLECTION = 'messages'\nDEFAULT_ROUTING_COLLECTION = 'messages.routing'\nDEFAULT_BROADCAST_COLLECTION = 'messages.broadcast'\n\n\nclass BroadcastCursor(object):\n    \"\"\"Cursor for broadcast queues.\"\"\"\n\n    def __init__(self, cursor):\n        self._cursor = cursor\n\n        self.purge(rewind=False)\n\n    def get_size(self):\n        return self._cursor.count() - self._offset\n\n    def close(self):\n        self._cursor.close()\n\n    def purge(self, rewind=True):\n        if rewind:\n            self._cursor.rewind()\n\n        # Fast forward the cursor past old events\n        self._offset = self._cursor.count()\n        self._cursor = self._cursor.skip(self._offset)\n\n    def __iter__(self):\n        return self\n\n    def __next__(self):\n        while True:\n            try:\n                msg = next(self._cursor)\n            except pymongo.errors.OperationFailure as exc:\n                # In some cases tailed cursor can become invalid\n                # and have to be reinitalized\n                if 'not valid at server' in exc.message:\n                    self.purge()\n\n                    continue\n\n                raise\n            else:\n                break\n\n        self._offset += 1\n\n        return msg\n    next = __next__\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(Channel, self).__init__(*vargs, **kwargs)\n\n        self._broadcast_cursors = {}\n\n        # Evaluate connection\n        self._create_client()\n\n    def _new_queue(self, queue, **kwargs):\n        pass\n\n    def _get(self, queue):\n        if queue in self._fanout_queues:\n            try:\n                msg = next(self.get_broadcast_cursor(queue))\n            except StopIteration:\n                msg = None\n        else:\n            msg = self.get_messages().find_and_modify(\n                query={'queue': queue},\n                sort=[('priority', pymongo.ASCENDING),\n                      ('_id', pymongo.ASCENDING)],\n                remove=True,\n            )\n\n        if msg is None:\n            raise Empty()\n\n        return loads(bytes_to_str(msg['payload']))\n\n    def _size(self, queue):\n        if queue in self._fanout_queues:\n            return self.get_broadcast_cursor(queue).get_size()\n\n        return self.get_messages().find({'queue': queue}).count()\n\n    def _put(self, queue, message, **kwargs):\n        self.get_messages().insert({\n            'payload': dumps(message),\n            'queue': queue,\n            'priority': self._get_message_priority(message, reverse=True),\n        })\n\n    def _purge(self, queue):\n        size = self._size(queue)\n\n        if queue in self._fanout_queues:\n            self.get_broadcaset_cursor(queue).purge()\n        else:\n            self.get_messages().remove({'queue': queue})\n\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 = {\n            'auto_start_request': True,\n            'ssl': client.ssl,\n            'connectTimeoutMS': (int(client.connect_timeout * 1000)\n                                 if client.connect_timeout else None),\n        }\n        options.update(client.transport_options)\n        options.update(parsed['options'])\n\n        return hostname, dbname, options\n\n    def _prepare_client_options(self, options):\n        if pymongo.version_tuple >= (3, ):\n            options.pop('auto_start_request', None)\n        return options\n\n    def _open(self, scheme='mongodb://'):\n        hostname, dbname, options = self._parse_uri(scheme=scheme)\n\n        conf = self._prepare_client_options(options)\n        conf['host'] = hostname\n\n        env = _detect_environment()\n        if env == 'gevent':\n            from gevent import monkey\n            monkey.patch_all()\n        elif env == 'eventlet':\n            from eventlet import monkey_patch\n            monkey_patch()\n\n        mongoconn = MongoClient(**conf)\n        database = 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._create_broadcast(database, options)\n\n        self._client = database\n\n    def _create_broadcast(self, database, options):\n        '''Create capped collection for broadcast messages.'''\n        if DEFAULT_BROADCAST_COLLECTION in database.collection_names():\n            return\n\n        capsize = options.get('capped_queue_size') or 100000\n        database.create_collection(DEFAULT_BROADCAST_COLLECTION,\n                                   size=capsize, capped=True)\n\n    def _ensure_indexes(self):\n        '''Ensure indexes on collections.'''\n        self.get_messages().ensure_index(\n            [('queue', 1), ('priority', 1), ('_id', 1)], background=True,\n        )\n        self.get_broadcast().ensure_index([('queue', 1)])\n        self.get_routing().ensure_index([('queue', 1), ('exchange', 1)])\n\n    def get_table(self, exchange):\n        \"\"\"Get table of bindings for ``exchange``.\"\"\"\n        # TODO Store a more complete exchange metatable in the\n        #      routing collection\n        localRoutes = frozenset(self.state.exchanges[exchange]['table'])\n        brokerRoutes = self.get_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, routing_key, **kwargs):\n        \"\"\"Deliver fanout message.\"\"\"\n        self.get_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            self.create_broadcast_cursor(exchange, routing_key, pattern, queue)\n            self._fanout_queues[queue] = exchange\n\n        meta = {'exchange': exchange,\n                'queue': queue,\n                'routing_key': routing_key,\n                'pattern': pattern}\n        self.get_routing().update(meta, meta, upsert=True)\n\n    def queue_delete(self, queue, **kwargs):\n        self.get_routing().remove({'queue': queue})\n\n        super(Channel, self).queue_delete(queue, **kwargs)\n\n        if queue in self._fanout_queues:\n            try:\n                cursor = self._broadcast_cursors.pop(queue)\n            except KeyError:\n                pass\n            else:\n                cursor.close()\n\n                self._fanout_queues.pop(queue)\n\n    def _create_client(self):\n        self._open()\n        self._ensure_indexes()\n\n    @property\n    def client(self):\n        if self._client is None:\n            self._create_client()\n        return self._client\n\n    def get_messages(self):\n        return self.client[DEFAULT_MESSAGES_COLLECTION]\n\n    def get_routing(self):\n        return self.client[DEFAULT_ROUTING_COLLECTION]\n\n    def get_broadcast(self):\n        return self.client[DEFAULT_BROADCAST_COLLECTION]\n\n    def get_broadcast_cursor(self, queue):\n        try:\n            return self._broadcast_cursors[queue]\n        except KeyError:\n            # Cursor may be absent when Channel created more than once.\n            # _fanout_queues is a class-level mutable attribute so it's\n            # shared over all Channel instances.\n            return self.create_broadcast_cursor(\n                self._fanout_queues[queue], None, None, queue,\n            )\n\n    def create_broadcast_cursor(self, exchange, routing_key, pattern, queue):\n        if pymongo.version_tuple >= (3, ):\n            query = dict(filter={'queue': exchange},\n                         sort=[('$natural', 1)],\n                         cursor_type=CursorType.TAILABLE\n                         )\n        else:\n            query = dict(query={'queue': exchange},\n                         sort=[('$natural', 1)],\n                         tailable=True\n                         )\n\n        cursor = self.get_broadcast().find(**query)\n        ret = self._broadcast_cursors[queue] = BroadcastCursor(cursor)\n        return ret\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    implements = virtual.Transport.implements.extend(\n        exchange_types=frozenset(['direct', 'topic', 'fanout']),\n    )\n\n    def driver_version(self):\n        return pymongo.version\n", "path": "kombu/transport/mongodb.py"}]}