We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
MongoDB and BROKER_USE_SSL=True
I've recently started with mongodb and BROKER_USE_SSL=True, this doesn't seem to work. Celery is trying to reconnect with Re-establishing connection message. BROKER_USE_SSL=False works well. 

``` python
[2013-02-21 14:57:45,708: DEBUG/MainProcess] consumer: Re-establishing connection to the broker...
[2013-02-21 14:57:45,710: INFO/MainProcess] consumer: Connected to mongodb://localhost%3A27017%2Fdata/data.
[2013-02-21 14:57:45,714: ERROR/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
Traceback (most recent call last):
  File "/stuff/eggs/celery-3.0.13-py2.7.egg/celery/worker/consumer.py", line 392, in start
    self.reset_connection()
  File "/stuff/eggs/celery-3.0.13-py2.7.egg/celery/worker/consumer.py", line 741, in reset_connection
    self.connection, on_decode_error=self.on_decode_error,
  File "/stuff/eggs/celery-3.0.13-py2.7.egg/celery/app/amqp.py", line 291, in __init__
    queues or self.app.amqp.queues.consume_from.values(), **kw
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/messaging.py", line 338, in __init__
    self.revive(self.channel)
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/messaging.py", line 350, in revive
    self.declare()
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/messaging.py", line 360, in declare
    queue.declare()
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/entity.py", line 471, in declare
    self.queue_declare(nowait, passive=False)
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/entity.py", line 497, in queue_declare
    nowait=nowait)
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/virtual/__init__.py", line 398, in queue_declare
    return queue, self._size(queue), 0
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/mongodb.py", line 76, in _size
    return self.client.messages.find({'queue': queue}).count()
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/mongodb.py", line 204, in client
    self._client = self._open()
  File "/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/mongodb.py", line 133, in _open
    mongoconn = Connection(host=hostname)
  File "/stuff/eggs/pymongo-2.4.2-py2.7-linux-x86_64.egg/pymongo/connection.py", line 180, in __init__
    max_pool_size, document_class, tz_aware, _connect, **kwargs)
  File "/stuff/eggs/pymongo-2.4.2-py2.7-linux-x86_64.egg/pymongo/mongo_client.py", line 269, in __init__
    raise ConnectionFailure(str(e))
ConnectionFailure: [Errno 104] Connection reset by peer
[2013-02-21 14:57:45,716: DEBUG/MainProcess] consumer: Re-establishing connection to the broker...
[2013-02-21 14:57:45,718: INFO/MainProcess] consumer: Connected to mongodb://localhost%3A27017%2Fdata/data.
[2013-02-21 14:57:45,721: ERROR/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...
```

Problem seems to be generated by this line 
https://github.com/celery/kombu/blob/master/kombu/transport/mongodb.py#L135
which should take ssl=True parameter for SSL connections.

I know it's kombu component, but setting (BROKER_USE_SSL) which is leading to this problem is part of celery library.  


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `celery/backends/mongodb.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 """
3     celery.backends.mongodb
4     ~~~~~~~~~~~~~~~~~~~~~~~
5 
6     MongoDB result store backend.
7 
8 """
9 from __future__ import absolute_import
10 
11 from datetime import datetime
12 
13 try:
14     import pymongo
15 except ImportError:  # pragma: no cover
16     pymongo = None   # noqa
17 
18 if pymongo:
19     try:
20         from bson.binary import Binary
21     except ImportError:                     # pragma: no cover
22         from pymongo.binary import Binary   # noqa
23 else:                                       # pragma: no cover
24     Binary = None                           # noqa
25 
26 from kombu.utils import cached_property
27 
28 from celery import states
29 from celery.exceptions import ImproperlyConfigured
30 from celery.five import string_t
31 from celery.utils.timeutils import maybe_timedelta
32 
33 from .base import BaseBackend
34 
35 
36 class Bunch(object):
37 
38     def __init__(self, **kw):
39         self.__dict__.update(kw)
40 
41 
42 class MongoBackend(BaseBackend):
43     mongodb_host = 'localhost'
44     mongodb_port = 27017
45     mongodb_user = None
46     mongodb_password = None
47     mongodb_database = 'celery'
48     mongodb_taskmeta_collection = 'celery_taskmeta'
49     mongodb_max_pool_size = 10
50 
51     def __init__(self, *args, **kwargs):
52         """Initialize MongoDB backend instance.
53 
54         :raises celery.exceptions.ImproperlyConfigured: if
55             module :mod:`pymongo` is not available.
56 
57         """
58         super(MongoBackend, self).__init__(*args, **kwargs)
59         self.expires = kwargs.get('expires') or maybe_timedelta(
60             self.app.conf.CELERY_TASK_RESULT_EXPIRES)
61 
62         if not pymongo:
63             raise ImproperlyConfigured(
64                 'You need to install the pymongo library to use the '
65                 'MongoDB backend.')
66 
67         config = self.app.conf.get('CELERY_MONGODB_BACKEND_SETTINGS', None)
68         if config is not None:
69             if not isinstance(config, dict):
70                 raise ImproperlyConfigured(
71                     'MongoDB backend settings should be grouped in a dict')
72 
73             self.mongodb_host = config.get('host', self.mongodb_host)
74             self.mongodb_port = int(config.get('port', self.mongodb_port))
75             self.mongodb_user = config.get('user', self.mongodb_user)
76             self.mongodb_password = config.get(
77                 'password', self.mongodb_password)
78             self.mongodb_database = config.get(
79                 'database', self.mongodb_database)
80             self.mongodb_taskmeta_collection = config.get(
81                 'taskmeta_collection', self.mongodb_taskmeta_collection)
82             self.mongodb_max_pool_size = config.get(
83                 'max_pool_size', self.mongodb_max_pool_size)
84 
85         self._connection = None
86 
87     def _get_connection(self):
88         """Connect to the MongoDB server."""
89         if self._connection is None:
90             from pymongo.connection import Connection
91 
92             # The first pymongo.Connection() argument (host) can be
93             # a list of ['host:port'] elements or a mongodb connection
94             # URI. If this is the case, don't use self.mongodb_port
95             # but let pymongo get the port(s) from the URI instead.
96             # This enables the use of replica sets and sharding.
97             # See pymongo.Connection() for more info.
98             args = [self.mongodb_host]
99             kwargs = {'max_pool_size': self.mongodb_max_pool_size}
100             if isinstance(self.mongodb_host, string_t) \
101                     and not self.mongodb_host.startswith('mongodb://'):
102                 args.append(self.mongodb_port)
103 
104             self._connection = Connection(*args, **kwargs)
105 
106         return self._connection
107 
108     def process_cleanup(self):
109         if self._connection is not None:
110             # MongoDB connection will be closed automatically when object
111             # goes out of scope
112             self._connection = None
113 
114     def _store_result(self, task_id, result, status, traceback=None):
115         """Store return value and status of an executed task."""
116         meta = {'_id': task_id,
117                 'status': status,
118                 'result': Binary(self.encode(result)),
119                 'date_done': datetime.utcnow(),
120                 'traceback': Binary(self.encode(traceback)),
121                 'children': Binary(self.encode(self.current_task_children()))}
122         self.collection.save(meta, safe=True)
123 
124         return result
125 
126     def _get_task_meta_for(self, task_id):
127         """Get task metadata for a task by id."""
128 
129         obj = self.collection.find_one({'_id': task_id})
130         if not obj:
131             return {'status': states.PENDING, 'result': None}
132 
133         meta = {
134             'task_id': obj['_id'],
135             'status': obj['status'],
136             'result': self.decode(obj['result']),
137             'date_done': obj['date_done'],
138             'traceback': self.decode(obj['traceback']),
139             'children': self.decode(obj['children']),
140         }
141 
142         return meta
143 
144     def _save_group(self, group_id, result):
145         """Save the group result."""
146         meta = {'_id': group_id,
147                 'result': Binary(self.encode(result)),
148                 'date_done': datetime.utcnow()}
149         self.collection.save(meta, safe=True)
150 
151         return result
152 
153     def _restore_group(self, group_id):
154         """Get the result for a group by id."""
155         obj = self.collection.find_one({'_id': group_id})
156         if not obj:
157             return
158 
159         meta = {
160             'task_id': obj['_id'],
161             'result': self.decode(obj['result']),
162             'date_done': obj['date_done'],
163         }
164 
165         return meta
166 
167     def _delete_group(self, group_id):
168         """Delete a group by id."""
169         self.collection.remove({'_id': group_id})
170 
171     def _forget(self, task_id):
172         """
173         Remove result from MongoDB.
174 
175         :raises celery.exceptions.OperationsError: if the task_id could not be
176                                                    removed.
177         """
178         # By using safe=True, this will wait until it receives a response from
179         # the server.  Likewise, it will raise an OperationsError if the
180         # response was unable to be completed.
181         self.collection.remove({'_id': task_id}, safe=True)
182 
183     def cleanup(self):
184         """Delete expired metadata."""
185         self.collection.remove(
186             {'date_done': {'$lt': self.app.now() - self.expires}},
187         )
188 
189     def __reduce__(self, args=(), kwargs={}):
190         kwargs.update(
191             dict(expires=self.expires))
192         return super(MongoBackend, self).__reduce__(args, kwargs)
193 
194     def _get_database(self):
195         conn = self._get_connection()
196         db = conn[self.mongodb_database]
197         if self.mongodb_user and self.mongodb_password:
198             if not db.authenticate(self.mongodb_user,
199                                    self.mongodb_password):
200                 raise ImproperlyConfigured(
201                     'Invalid MongoDB username or password.')
202         return db
203 
204     @cached_property
205     def database(self):
206         """Get database from MongoDB connection and perform authentication
207         if necessary."""
208         return self._get_database()
209 
210     @cached_property
211     def collection(self):
212         """Get the metadata task collection."""
213         collection = self.database[self.mongodb_taskmeta_collection]
214 
215         # Ensure an index on date_done is there, if not process the index
216         # in the background. Once completed cleanup will be much faster
217         collection.ensure_index('date_done', background='true')
218         return collection
219
```
--- 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/celery/backends/mongodb.py b/celery/backends/mongodb.py
--- a/celery/backends/mongodb.py
+++ b/celery/backends/mongodb.py
@@ -96,7 +96,10 @@
             # This enables the use of replica sets and sharding.
             # See pymongo.Connection() for more info.
             args = [self.mongodb_host]
-            kwargs = {'max_pool_size': self.mongodb_max_pool_size}
+            kwargs = {
+                'max_pool_size': self.mongodb_max_pool_size,
+                'ssl': self.app.conf.BROKER_USE_SSL
+            }
             if isinstance(self.mongodb_host, string_t) \
                     and not self.mongodb_host.startswith('mongodb://'):
                 args.append(self.mongodb_port)

{"golden_diff": "diff --git a/celery/backends/mongodb.py b/celery/backends/mongodb.py\n--- a/celery/backends/mongodb.py\n+++ b/celery/backends/mongodb.py\n@@ -96,7 +96,10 @@\n             # This enables the use of replica sets and sharding.\n             # See pymongo.Connection() for more info.\n             args = [self.mongodb_host]\n-            kwargs = {'max_pool_size': self.mongodb_max_pool_size}\n+            kwargs = {\n+                'max_pool_size': self.mongodb_max_pool_size,\n+                'ssl': self.app.conf.BROKER_USE_SSL\n+            }\n             if isinstance(self.mongodb_host, string_t) \\\n                     and not self.mongodb_host.startswith('mongodb://'):\n                 args.append(self.mongodb_port)\n", "issue": "MongoDB and BROKER_USE_SSL=True\nI've recently started with mongodb and BROKER_USE_SSL=True, this doesn't seem to work. Celery is trying to reconnect with Re-establishing connection message. BROKER_USE_SSL=False works well. \n\n``` python\n[2013-02-21 14:57:45,708: DEBUG/MainProcess] consumer: Re-establishing connection to the broker...\n[2013-02-21 14:57:45,710: INFO/MainProcess] consumer: Connected to mongodb://localhost%3A27017%2Fdata/data.\n[2013-02-21 14:57:45,714: ERROR/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...\nTraceback (most recent call last):\n  File \"/stuff/eggs/celery-3.0.13-py2.7.egg/celery/worker/consumer.py\", line 392, in start\n    self.reset_connection()\n  File \"/stuff/eggs/celery-3.0.13-py2.7.egg/celery/worker/consumer.py\", line 741, in reset_connection\n    self.connection, on_decode_error=self.on_decode_error,\n  File \"/stuff/eggs/celery-3.0.13-py2.7.egg/celery/app/amqp.py\", line 291, in __init__\n    queues or self.app.amqp.queues.consume_from.values(), **kw\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/messaging.py\", line 338, in __init__\n    self.revive(self.channel)\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/messaging.py\", line 350, in revive\n    self.declare()\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/messaging.py\", line 360, in declare\n    queue.declare()\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/entity.py\", line 471, in declare\n    self.queue_declare(nowait, passive=False)\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/entity.py\", line 497, in queue_declare\n    nowait=nowait)\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/virtual/__init__.py\", line 398, in queue_declare\n    return queue, self._size(queue), 0\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/mongodb.py\", line 76, in _size\n    return self.client.messages.find({'queue': queue}).count()\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/mongodb.py\", line 204, in client\n    self._client = self._open()\n  File \"/stuff/eggs/kombu-2.5.4-py2.7.egg/kombu/transport/mongodb.py\", line 133, in _open\n    mongoconn = Connection(host=hostname)\n  File \"/stuff/eggs/pymongo-2.4.2-py2.7-linux-x86_64.egg/pymongo/connection.py\", line 180, in __init__\n    max_pool_size, document_class, tz_aware, _connect, **kwargs)\n  File \"/stuff/eggs/pymongo-2.4.2-py2.7-linux-x86_64.egg/pymongo/mongo_client.py\", line 269, in __init__\n    raise ConnectionFailure(str(e))\nConnectionFailure: [Errno 104] Connection reset by peer\n[2013-02-21 14:57:45,716: DEBUG/MainProcess] consumer: Re-establishing connection to the broker...\n[2013-02-21 14:57:45,718: INFO/MainProcess] consumer: Connected to mongodb://localhost%3A27017%2Fdata/data.\n[2013-02-21 14:57:45,721: ERROR/MainProcess] consumer: Connection to broker lost. Trying to re-establish the connection...\n```\n\nProblem seems to be generated by this line \nhttps://github.com/celery/kombu/blob/master/kombu/transport/mongodb.py#L135\nwhich should take ssl=True parameter for SSL connections.\n\nI know it's kombu component, but setting (BROKER_USE_SSL) which is leading to this problem is part of celery library.  \n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\n    celery.backends.mongodb\n    ~~~~~~~~~~~~~~~~~~~~~~~\n\n    MongoDB result store backend.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nfrom datetime import datetime\n\ntry:\n    import pymongo\nexcept ImportError:  # pragma: no cover\n    pymongo = None   # noqa\n\nif pymongo:\n    try:\n        from bson.binary import Binary\n    except ImportError:                     # pragma: no cover\n        from pymongo.binary import Binary   # noqa\nelse:                                       # pragma: no cover\n    Binary = None                           # noqa\n\nfrom kombu.utils import cached_property\n\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import string_t\nfrom celery.utils.timeutils import maybe_timedelta\n\nfrom .base import BaseBackend\n\n\nclass Bunch(object):\n\n    def __init__(self, **kw):\n        self.__dict__.update(kw)\n\n\nclass MongoBackend(BaseBackend):\n    mongodb_host = 'localhost'\n    mongodb_port = 27017\n    mongodb_user = None\n    mongodb_password = None\n    mongodb_database = 'celery'\n    mongodb_taskmeta_collection = 'celery_taskmeta'\n    mongodb_max_pool_size = 10\n\n    def __init__(self, *args, **kwargs):\n        \"\"\"Initialize MongoDB backend instance.\n\n        :raises celery.exceptions.ImproperlyConfigured: if\n            module :mod:`pymongo` is not available.\n\n        \"\"\"\n        super(MongoBackend, self).__init__(*args, **kwargs)\n        self.expires = kwargs.get('expires') or maybe_timedelta(\n            self.app.conf.CELERY_TASK_RESULT_EXPIRES)\n\n        if not pymongo:\n            raise ImproperlyConfigured(\n                'You need to install the pymongo library to use the '\n                'MongoDB backend.')\n\n        config = self.app.conf.get('CELERY_MONGODB_BACKEND_SETTINGS', None)\n        if config is not None:\n            if not isinstance(config, dict):\n                raise ImproperlyConfigured(\n                    'MongoDB backend settings should be grouped in a dict')\n\n            self.mongodb_host = config.get('host', self.mongodb_host)\n            self.mongodb_port = int(config.get('port', self.mongodb_port))\n            self.mongodb_user = config.get('user', self.mongodb_user)\n            self.mongodb_password = config.get(\n                'password', self.mongodb_password)\n            self.mongodb_database = config.get(\n                'database', self.mongodb_database)\n            self.mongodb_taskmeta_collection = config.get(\n                'taskmeta_collection', self.mongodb_taskmeta_collection)\n            self.mongodb_max_pool_size = config.get(\n                'max_pool_size', self.mongodb_max_pool_size)\n\n        self._connection = None\n\n    def _get_connection(self):\n        \"\"\"Connect to the MongoDB server.\"\"\"\n        if self._connection is None:\n            from pymongo.connection import Connection\n\n            # The first pymongo.Connection() argument (host) can be\n            # a list of ['host:port'] elements or a mongodb connection\n            # URI. If this is the case, don't use self.mongodb_port\n            # but let pymongo get the port(s) from the URI instead.\n            # This enables the use of replica sets and sharding.\n            # See pymongo.Connection() for more info.\n            args = [self.mongodb_host]\n            kwargs = {'max_pool_size': self.mongodb_max_pool_size}\n            if isinstance(self.mongodb_host, string_t) \\\n                    and not self.mongodb_host.startswith('mongodb://'):\n                args.append(self.mongodb_port)\n\n            self._connection = Connection(*args, **kwargs)\n\n        return self._connection\n\n    def process_cleanup(self):\n        if self._connection is not None:\n            # MongoDB connection will be closed automatically when object\n            # goes out of scope\n            self._connection = None\n\n    def _store_result(self, task_id, result, status, traceback=None):\n        \"\"\"Store return value and status of an executed task.\"\"\"\n        meta = {'_id': task_id,\n                'status': status,\n                'result': Binary(self.encode(result)),\n                'date_done': datetime.utcnow(),\n                'traceback': Binary(self.encode(traceback)),\n                'children': Binary(self.encode(self.current_task_children()))}\n        self.collection.save(meta, safe=True)\n\n        return result\n\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task metadata for a task by id.\"\"\"\n\n        obj = self.collection.find_one({'_id': task_id})\n        if not obj:\n            return {'status': states.PENDING, 'result': None}\n\n        meta = {\n            'task_id': obj['_id'],\n            'status': obj['status'],\n            'result': self.decode(obj['result']),\n            'date_done': obj['date_done'],\n            'traceback': self.decode(obj['traceback']),\n            'children': self.decode(obj['children']),\n        }\n\n        return meta\n\n    def _save_group(self, group_id, result):\n        \"\"\"Save the group result.\"\"\"\n        meta = {'_id': group_id,\n                'result': Binary(self.encode(result)),\n                'date_done': datetime.utcnow()}\n        self.collection.save(meta, safe=True)\n\n        return result\n\n    def _restore_group(self, group_id):\n        \"\"\"Get the result for a group by id.\"\"\"\n        obj = self.collection.find_one({'_id': group_id})\n        if not obj:\n            return\n\n        meta = {\n            'task_id': obj['_id'],\n            'result': self.decode(obj['result']),\n            'date_done': obj['date_done'],\n        }\n\n        return meta\n\n    def _delete_group(self, group_id):\n        \"\"\"Delete a group by id.\"\"\"\n        self.collection.remove({'_id': group_id})\n\n    def _forget(self, task_id):\n        \"\"\"\n        Remove result from MongoDB.\n\n        :raises celery.exceptions.OperationsError: if the task_id could not be\n                                                   removed.\n        \"\"\"\n        # By using safe=True, this will wait until it receives a response from\n        # the server.  Likewise, it will raise an OperationsError if the\n        # response was unable to be completed.\n        self.collection.remove({'_id': task_id}, safe=True)\n\n    def cleanup(self):\n        \"\"\"Delete expired metadata.\"\"\"\n        self.collection.remove(\n            {'date_done': {'$lt': self.app.now() - self.expires}},\n        )\n\n    def __reduce__(self, args=(), kwargs={}):\n        kwargs.update(\n            dict(expires=self.expires))\n        return super(MongoBackend, self).__reduce__(args, kwargs)\n\n    def _get_database(self):\n        conn = self._get_connection()\n        db = conn[self.mongodb_database]\n        if self.mongodb_user and self.mongodb_password:\n            if not db.authenticate(self.mongodb_user,\n                                   self.mongodb_password):\n                raise ImproperlyConfigured(\n                    'Invalid MongoDB username or password.')\n        return db\n\n    @cached_property\n    def database(self):\n        \"\"\"Get database from MongoDB connection and perform authentication\n        if necessary.\"\"\"\n        return self._get_database()\n\n    @cached_property\n    def collection(self):\n        \"\"\"Get the metadata task collection.\"\"\"\n        collection = self.database[self.mongodb_taskmeta_collection]\n\n        # Ensure an index on date_done is there, if not process the index\n        # in the background. Once completed cleanup will be much faster\n        collection.ensure_index('date_done', background='true')\n        return collection\n", "path": "celery/backends/mongodb.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\n    celery.backends.mongodb\n    ~~~~~~~~~~~~~~~~~~~~~~~\n\n    MongoDB result store backend.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nfrom datetime import datetime\n\ntry:\n    import pymongo\nexcept ImportError:  # pragma: no cover\n    pymongo = None   # noqa\n\nif pymongo:\n    try:\n        from bson.binary import Binary\n    except ImportError:                     # pragma: no cover\n        from pymongo.binary import Binary   # noqa\nelse:                                       # pragma: no cover\n    Binary = None                           # noqa\n\nfrom kombu.utils import cached_property\n\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.five import string_t\nfrom celery.utils.timeutils import maybe_timedelta\n\nfrom .base import BaseBackend\n\n\nclass Bunch(object):\n\n    def __init__(self, **kw):\n        self.__dict__.update(kw)\n\n\nclass MongoBackend(BaseBackend):\n    mongodb_host = 'localhost'\n    mongodb_port = 27017\n    mongodb_user = None\n    mongodb_password = None\n    mongodb_database = 'celery'\n    mongodb_taskmeta_collection = 'celery_taskmeta'\n    mongodb_max_pool_size = 10\n\n    def __init__(self, *args, **kwargs):\n        \"\"\"Initialize MongoDB backend instance.\n\n        :raises celery.exceptions.ImproperlyConfigured: if\n            module :mod:`pymongo` is not available.\n\n        \"\"\"\n        super(MongoBackend, self).__init__(*args, **kwargs)\n        self.expires = kwargs.get('expires') or maybe_timedelta(\n            self.app.conf.CELERY_TASK_RESULT_EXPIRES)\n\n        if not pymongo:\n            raise ImproperlyConfigured(\n                'You need to install the pymongo library to use the '\n                'MongoDB backend.')\n\n        config = self.app.conf.get('CELERY_MONGODB_BACKEND_SETTINGS', None)\n        if config is not None:\n            if not isinstance(config, dict):\n                raise ImproperlyConfigured(\n                    'MongoDB backend settings should be grouped in a dict')\n\n            self.mongodb_host = config.get('host', self.mongodb_host)\n            self.mongodb_port = int(config.get('port', self.mongodb_port))\n            self.mongodb_user = config.get('user', self.mongodb_user)\n            self.mongodb_password = config.get(\n                'password', self.mongodb_password)\n            self.mongodb_database = config.get(\n                'database', self.mongodb_database)\n            self.mongodb_taskmeta_collection = config.get(\n                'taskmeta_collection', self.mongodb_taskmeta_collection)\n            self.mongodb_max_pool_size = config.get(\n                'max_pool_size', self.mongodb_max_pool_size)\n\n        self._connection = None\n\n    def _get_connection(self):\n        \"\"\"Connect to the MongoDB server.\"\"\"\n        if self._connection is None:\n            from pymongo.connection import Connection\n\n            # The first pymongo.Connection() argument (host) can be\n            # a list of ['host:port'] elements or a mongodb connection\n            # URI. If this is the case, don't use self.mongodb_port\n            # but let pymongo get the port(s) from the URI instead.\n            # This enables the use of replica sets and sharding.\n            # See pymongo.Connection() for more info.\n            args = [self.mongodb_host]\n            kwargs = {\n                'max_pool_size': self.mongodb_max_pool_size,\n                'ssl': self.app.conf.BROKER_USE_SSL\n            }\n            if isinstance(self.mongodb_host, string_t) \\\n                    and not self.mongodb_host.startswith('mongodb://'):\n                args.append(self.mongodb_port)\n\n            self._connection = Connection(*args, **kwargs)\n\n        return self._connection\n\n    def process_cleanup(self):\n        if self._connection is not None:\n            # MongoDB connection will be closed automatically when object\n            # goes out of scope\n            self._connection = None\n\n    def _store_result(self, task_id, result, status, traceback=None):\n        \"\"\"Store return value and status of an executed task.\"\"\"\n        meta = {'_id': task_id,\n                'status': status,\n                'result': Binary(self.encode(result)),\n                'date_done': datetime.utcnow(),\n                'traceback': Binary(self.encode(traceback)),\n                'children': Binary(self.encode(self.current_task_children()))}\n        self.collection.save(meta, safe=True)\n\n        return result\n\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task metadata for a task by id.\"\"\"\n\n        obj = self.collection.find_one({'_id': task_id})\n        if not obj:\n            return {'status': states.PENDING, 'result': None}\n\n        meta = {\n            'task_id': obj['_id'],\n            'status': obj['status'],\n            'result': self.decode(obj['result']),\n            'date_done': obj['date_done'],\n            'traceback': self.decode(obj['traceback']),\n            'children': self.decode(obj['children']),\n        }\n\n        return meta\n\n    def _save_group(self, group_id, result):\n        \"\"\"Save the group result.\"\"\"\n        meta = {'_id': group_id,\n                'result': Binary(self.encode(result)),\n                'date_done': datetime.utcnow()}\n        self.collection.save(meta, safe=True)\n\n        return result\n\n    def _restore_group(self, group_id):\n        \"\"\"Get the result for a group by id.\"\"\"\n        obj = self.collection.find_one({'_id': group_id})\n        if not obj:\n            return\n\n        meta = {\n            'task_id': obj['_id'],\n            'result': self.decode(obj['result']),\n            'date_done': obj['date_done'],\n        }\n\n        return meta\n\n    def _delete_group(self, group_id):\n        \"\"\"Delete a group by id.\"\"\"\n        self.collection.remove({'_id': group_id})\n\n    def _forget(self, task_id):\n        \"\"\"\n        Remove result from MongoDB.\n\n        :raises celery.exceptions.OperationsError: if the task_id could not be\n                                                   removed.\n        \"\"\"\n        # By using safe=True, this will wait until it receives a response from\n        # the server.  Likewise, it will raise an OperationsError if the\n        # response was unable to be completed.\n        self.collection.remove({'_id': task_id}, safe=True)\n\n    def cleanup(self):\n        \"\"\"Delete expired metadata.\"\"\"\n        self.collection.remove(\n            {'date_done': {'$lt': self.app.now() - self.expires}},\n        )\n\n    def __reduce__(self, args=(), kwargs={}):\n        kwargs.update(\n            dict(expires=self.expires))\n        return super(MongoBackend, self).__reduce__(args, kwargs)\n\n    def _get_database(self):\n        conn = self._get_connection()\n        db = conn[self.mongodb_database]\n        if self.mongodb_user and self.mongodb_password:\n            if not db.authenticate(self.mongodb_user,\n                                   self.mongodb_password):\n                raise ImproperlyConfigured(\n                    'Invalid MongoDB username or password.')\n        return db\n\n    @cached_property\n    def database(self):\n        \"\"\"Get database from MongoDB connection and perform authentication\n        if necessary.\"\"\"\n        return self._get_database()\n\n    @cached_property\n    def collection(self):\n        \"\"\"Get the metadata task collection.\"\"\"\n        collection = self.database[self.mongodb_taskmeta_collection]\n\n        # Ensure an index on date_done is there, if not process the index\n        # in the background. Once completed cleanup will be much faster\n        collection.ensure_index('date_done', background='true')\n        return collection\n", "path": "celery/backends/mongodb.py"}]}