We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
CELERY_RESULT_SERIALIZER = 'json' breaks Exception marshaling
Setting `CELERY_RESULT_SERIALIZER = json` and raising an exception in the worker leads to this:

```
/path/to/lib/python2.7/site-packages/celery/result.py in get(self, timeout, propagate, interval, no_ack, follow_parents, EXCEPTION_STATES, PROPAGATE_STATES)
    173             status = meta['status']
    174             if status in PROPAGATE_STATES and propagate:
--> 175                 raise meta['result']
    176             return meta['result']
    177     wait = get  # deprecated alias to :meth:`get`.

TypeError: exceptions must be old-style classes or derived from BaseException, not dict
```

where the contents of `meta['result']` are (in my case):

```
{u'exc_message': u'unknown keys: nam', u'exc_type': u'ValueError'}
```

so it _looks_ like celery could convert the dict to a real exception before raising, but it does not currently.  Changing back to `pickle` works as expected.

bug can be reproduced with the following:

``` python
# jsonresults.py
from celery.app.base import Celery

CELERY_RESULT_SERIALIZER = 'json'
CELERY_RESULT_BACKEND = 'amqp'

app = Celery(config_source=__name__)

@app.task
def hello():
    raise ValueError('go away')
```

worker:

```
# python -m celery --app=jsonresults:app worker
```

caller:

``` python
import jsonresults
jsonresults.hello.delay().get()
```


--- 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/amqp.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 """
3     celery.backends.amqp
4     ~~~~~~~~~~~~~~~~~~~~
5 
6     The AMQP result backend.
7 
8     This backend publishes results as messages.
9 
10 """
11 from __future__ import absolute_import
12 
13 import socket
14 
15 from collections import deque
16 from operator import itemgetter
17 
18 from kombu import Exchange, Queue, Producer, Consumer
19 
20 from celery import states
21 from celery.exceptions import TimeoutError
22 from celery.five import range, monotonic
23 from celery.utils.functional import dictfilter
24 from celery.utils.log import get_logger
25 from celery.utils.timeutils import maybe_s_to_ms
26 
27 from .base import BaseBackend
28 
29 __all__ = ['BacklogLimitExceeded', 'AMQPBackend']
30 
31 logger = get_logger(__name__)
32 
33 
34 class BacklogLimitExceeded(Exception):
35     """Too much state history to fast-forward."""
36 
37 
38 def repair_uuid(s):
39     # Historically the dashes in UUIDS are removed from AMQ entity names,
40     # but there is no known reason to.  Hopefully we'll be able to fix
41     # this in v4.0.
42     return '%s-%s-%s-%s-%s' % (s[:8], s[8:12], s[12:16], s[16:20], s[20:])
43 
44 
45 class NoCacheQueue(Queue):
46     can_cache_declaration = False
47 
48 
49 class AMQPBackend(BaseBackend):
50     """Publishes results by sending messages."""
51     Exchange = Exchange
52     Queue = NoCacheQueue
53     Consumer = Consumer
54     Producer = Producer
55 
56     BacklogLimitExceeded = BacklogLimitExceeded
57 
58     persistent = True
59     supports_autoexpire = True
60     supports_native_join = True
61 
62     retry_policy = {
63         'max_retries': 20,
64         'interval_start': 0,
65         'interval_step': 1,
66         'interval_max': 1,
67     }
68 
69     def __init__(self, app, connection=None, exchange=None, exchange_type=None,
70                  persistent=None, serializer=None, auto_delete=True, **kwargs):
71         super(AMQPBackend, self).__init__(app, **kwargs)
72         conf = self.app.conf
73         self._connection = connection
74         self.persistent = self.prepare_persistent(persistent)
75         self.delivery_mode = 2 if self.persistent else 1
76         exchange = exchange or conf.CELERY_RESULT_EXCHANGE
77         exchange_type = exchange_type or conf.CELERY_RESULT_EXCHANGE_TYPE
78         self.exchange = self._create_exchange(
79             exchange, exchange_type, self.delivery_mode,
80         )
81         self.serializer = serializer or conf.CELERY_RESULT_SERIALIZER
82         self.auto_delete = auto_delete
83         self.queue_arguments = dictfilter({
84             'x-expires': maybe_s_to_ms(self.expires),
85         })
86 
87     def _create_exchange(self, name, type='direct', delivery_mode=2):
88         return self.Exchange(name=name,
89                              type=type,
90                              delivery_mode=delivery_mode,
91                              durable=self.persistent,
92                              auto_delete=False)
93 
94     def _create_binding(self, task_id):
95         name = self.rkey(task_id)
96         return self.Queue(name=name,
97                           exchange=self.exchange,
98                           routing_key=name,
99                           durable=self.persistent,
100                           auto_delete=self.auto_delete,
101                           queue_arguments=self.queue_arguments)
102 
103     def revive(self, channel):
104         pass
105 
106     def rkey(self, task_id):
107         return task_id.replace('-', '')
108 
109     def destination_for(self, task_id, request):
110         if request:
111             return self.rkey(task_id), request.correlation_id or task_id
112         return self.rkey(task_id), task_id
113 
114     def store_result(self, task_id, result, status,
115                      traceback=None, request=None, **kwargs):
116         """Send task return value and status."""
117         routing_key, correlation_id = self.destination_for(task_id, request)
118         if not routing_key:
119             return
120         with self.app.amqp.producer_pool.acquire(block=True) as producer:
121             producer.publish(
122                 {'task_id': task_id, 'status': status,
123                  'result': self.encode_result(result, status),
124                  'traceback': traceback,
125                  'children': self.current_task_children(request)},
126                 exchange=self.exchange,
127                 routing_key=routing_key,
128                 correlation_id=correlation_id,
129                 serializer=self.serializer,
130                 retry=True, retry_policy=self.retry_policy,
131                 declare=self.on_reply_declare(task_id),
132                 delivery_mode=self.delivery_mode,
133             )
134         return result
135 
136     def on_reply_declare(self, task_id):
137         return [self._create_binding(task_id)]
138 
139     def wait_for(self, task_id, timeout=None, cache=True,
140                  no_ack=True, on_interval=None,
141                  READY_STATES=states.READY_STATES,
142                  PROPAGATE_STATES=states.PROPAGATE_STATES,
143                  **kwargs):
144         cached_meta = self._cache.get(task_id)
145         if cache and cached_meta and \
146                 cached_meta['status'] in READY_STATES:
147             return cached_meta
148         else:
149             try:
150                 return self.consume(task_id, timeout=timeout, no_ack=no_ack,
151                                     on_interval=on_interval)
152             except socket.timeout:
153                 raise TimeoutError('The operation timed out.')
154 
155     def get_task_meta(self, task_id, backlog_limit=1000):
156         # Polling and using basic_get
157         with self.app.pool.acquire_channel(block=True) as (_, channel):
158             binding = self._create_binding(task_id)(channel)
159             binding.declare()
160 
161             prev = latest = acc = None
162             for i in range(backlog_limit):  # spool ffwd
163                 acc = binding.get(
164                     accept=self.accept, no_ack=False,
165                 )
166                 if not acc:  # no more messages
167                     break
168                 if acc.payload['task_id'] == task_id:
169                     prev, latest = latest, acc
170                 if prev:
171                     # backends are not expected to keep history,
172                     # so we delete everything except the most recent state.
173                     prev.ack()
174                     prev = None
175             else:
176                 raise self.BacklogLimitExceeded(task_id)
177 
178             if latest:
179                 payload = self._cache[task_id] = latest.payload
180                 latest.requeue()
181                 return payload
182             else:
183                 # no new state, use previous
184                 try:
185                     return self._cache[task_id]
186                 except KeyError:
187                     # result probably pending.
188                     return {'status': states.PENDING, 'result': None}
189     poll = get_task_meta  # XXX compat
190 
191     def drain_events(self, connection, consumer,
192                      timeout=None, on_interval=None, now=monotonic, wait=None):
193         wait = wait or connection.drain_events
194         results = {}
195 
196         def callback(meta, message):
197             if meta['status'] in states.READY_STATES:
198                 results[meta['task_id']] = meta
199 
200         consumer.callbacks[:] = [callback]
201         time_start = now()
202 
203         while 1:
204             # Total time spent may exceed a single call to wait()
205             if timeout and now() - time_start >= timeout:
206                 raise socket.timeout()
207             try:
208                 wait(timeout=1)
209             except socket.timeout:
210                 pass
211             if on_interval:
212                 on_interval()
213             if results:  # got event on the wanted channel.
214                 break
215         self._cache.update(results)
216         return results
217 
218     def consume(self, task_id, timeout=None, no_ack=True, on_interval=None):
219         wait = self.drain_events
220         with self.app.pool.acquire_channel(block=True) as (conn, channel):
221             binding = self._create_binding(task_id)
222             with self.Consumer(channel, binding,
223                                no_ack=no_ack, accept=self.accept) as consumer:
224                 while 1:
225                     try:
226                         return wait(
227                             conn, consumer, timeout, on_interval)[task_id]
228                     except KeyError:
229                         continue
230 
231     def _many_bindings(self, ids):
232         return [self._create_binding(task_id) for task_id in ids]
233 
234     def get_many(self, task_ids, timeout=None, no_ack=True, on_message=None,
235                  now=monotonic, getfields=itemgetter('status', 'task_id'),
236                  READY_STATES=states.READY_STATES,
237                  PROPAGATE_STATES=states.PROPAGATE_STATES, **kwargs):
238         with self.app.pool.acquire_channel(block=True) as (conn, channel):
239             ids = set(task_ids)
240             cached_ids = set()
241             mark_cached = cached_ids.add
242             for task_id in ids:
243                 try:
244                     cached = self._cache[task_id]
245                 except KeyError:
246                     pass
247                 else:
248                     if cached['status'] in READY_STATES:
249                         yield task_id, cached
250                         mark_cached(task_id)
251             ids.difference_update(cached_ids)
252             results = deque()
253             push_result = results.append
254             push_cache = self._cache.__setitem__
255             decode_result = self.meta_from_decoded
256 
257             def _on_message(message):
258                 body = decode_result(message.decode())
259                 if on_message is not None:
260                     on_message(body)
261                 state, uid = getfields(body)
262                 if state in READY_STATES:
263                     push_result(body) \
264                         if uid in task_ids else push_cache(uid, body)
265 
266             bindings = self._many_bindings(task_ids)
267             with self.Consumer(channel, bindings, on_message=_on_message,
268                                accept=self.accept, no_ack=no_ack):
269                 wait = conn.drain_events
270                 popleft = results.popleft
271                 while ids:
272                     wait(timeout=timeout)
273                     while results:
274                         state = popleft()
275                         task_id = state['task_id']
276                         ids.discard(task_id)
277                         push_cache(task_id, state)
278                         yield task_id, state
279 
280     def reload_task_result(self, task_id):
281         raise NotImplementedError(
282             'reload_task_result is not supported by this backend.')
283 
284     def reload_group_result(self, task_id):
285         """Reload group result, even if it has been previously fetched."""
286         raise NotImplementedError(
287             'reload_group_result is not supported by this backend.')
288 
289     def save_group(self, group_id, result):
290         raise NotImplementedError(
291             'save_group is not supported by this backend.')
292 
293     def restore_group(self, group_id, cache=True):
294         raise NotImplementedError(
295             'restore_group is not supported by this backend.')
296 
297     def delete_group(self, group_id):
298         raise NotImplementedError(
299             'delete_group is not supported by this backend.')
300 
301     def __reduce__(self, args=(), kwargs={}):
302         kwargs.update(
303             connection=self._connection,
304             exchange=self.exchange.name,
305             exchange_type=self.exchange.type,
306             persistent=self.persistent,
307             serializer=self.serializer,
308             auto_delete=self.auto_delete,
309             expires=self.expires,
310         )
311         return super(AMQPBackend, self).__reduce__(args, kwargs)
312
```
--- 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/amqp.py b/celery/backends/amqp.py
--- a/celery/backends/amqp.py
+++ b/celery/backends/amqp.py
@@ -195,7 +195,7 @@
 
         def callback(meta, message):
             if meta['status'] in states.READY_STATES:
-                results[meta['task_id']] = meta
+                results[meta['task_id']] = self.meta_from_decoded(meta)
 
         consumer.callbacks[:] = [callback]
         time_start = now()

{"golden_diff": "diff --git a/celery/backends/amqp.py b/celery/backends/amqp.py\n--- a/celery/backends/amqp.py\n+++ b/celery/backends/amqp.py\n@@ -195,7 +195,7 @@\n \n         def callback(meta, message):\n             if meta['status'] in states.READY_STATES:\n-                results[meta['task_id']] = meta\n+                results[meta['task_id']] = self.meta_from_decoded(meta)\n \n         consumer.callbacks[:] = [callback]\n         time_start = now()\n", "issue": "CELERY_RESULT_SERIALIZER = 'json' breaks Exception marshaling\nSetting `CELERY_RESULT_SERIALIZER = json` and raising an exception in the worker leads to this:\n\n```\n/path/to/lib/python2.7/site-packages/celery/result.py in get(self, timeout, propagate, interval, no_ack, follow_parents, EXCEPTION_STATES, PROPAGATE_STATES)\n    173             status = meta['status']\n    174             if status in PROPAGATE_STATES and propagate:\n--> 175                 raise meta['result']\n    176             return meta['result']\n    177     wait = get  # deprecated alias to :meth:`get`.\n\nTypeError: exceptions must be old-style classes or derived from BaseException, not dict\n```\n\nwhere the contents of `meta['result']` are (in my case):\n\n```\n{u'exc_message': u'unknown keys: nam', u'exc_type': u'ValueError'}\n```\n\nso it _looks_ like celery could convert the dict to a real exception before raising, but it does not currently.  Changing back to `pickle` works as expected.\n\nbug can be reproduced with the following:\n\n``` python\n# jsonresults.py\nfrom celery.app.base import Celery\n\nCELERY_RESULT_SERIALIZER = 'json'\nCELERY_RESULT_BACKEND = 'amqp'\n\napp = Celery(config_source=__name__)\n\n@app.task\ndef hello():\n    raise ValueError('go away')\n```\n\nworker:\n\n```\n# python -m celery --app=jsonresults:app worker\n```\n\ncaller:\n\n``` python\nimport jsonresults\njsonresults.hello.delay().get()\n```\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\n    celery.backends.amqp\n    ~~~~~~~~~~~~~~~~~~~~\n\n    The AMQP result backend.\n\n    This backend publishes results as messages.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nimport socket\n\nfrom collections import deque\nfrom operator import itemgetter\n\nfrom kombu import Exchange, Queue, Producer, Consumer\n\nfrom celery import states\nfrom celery.exceptions import TimeoutError\nfrom celery.five import range, monotonic\nfrom celery.utils.functional import dictfilter\nfrom celery.utils.log import get_logger\nfrom celery.utils.timeutils import maybe_s_to_ms\n\nfrom .base import BaseBackend\n\n__all__ = ['BacklogLimitExceeded', 'AMQPBackend']\n\nlogger = get_logger(__name__)\n\n\nclass BacklogLimitExceeded(Exception):\n    \"\"\"Too much state history to fast-forward.\"\"\"\n\n\ndef repair_uuid(s):\n    # Historically the dashes in UUIDS are removed from AMQ entity names,\n    # but there is no known reason to.  Hopefully we'll be able to fix\n    # this in v4.0.\n    return '%s-%s-%s-%s-%s' % (s[:8], s[8:12], s[12:16], s[16:20], s[20:])\n\n\nclass NoCacheQueue(Queue):\n    can_cache_declaration = False\n\n\nclass AMQPBackend(BaseBackend):\n    \"\"\"Publishes results by sending messages.\"\"\"\n    Exchange = Exchange\n    Queue = NoCacheQueue\n    Consumer = Consumer\n    Producer = Producer\n\n    BacklogLimitExceeded = BacklogLimitExceeded\n\n    persistent = True\n    supports_autoexpire = True\n    supports_native_join = True\n\n    retry_policy = {\n        'max_retries': 20,\n        'interval_start': 0,\n        'interval_step': 1,\n        'interval_max': 1,\n    }\n\n    def __init__(self, app, connection=None, exchange=None, exchange_type=None,\n                 persistent=None, serializer=None, auto_delete=True, **kwargs):\n        super(AMQPBackend, self).__init__(app, **kwargs)\n        conf = self.app.conf\n        self._connection = connection\n        self.persistent = self.prepare_persistent(persistent)\n        self.delivery_mode = 2 if self.persistent else 1\n        exchange = exchange or conf.CELERY_RESULT_EXCHANGE\n        exchange_type = exchange_type or conf.CELERY_RESULT_EXCHANGE_TYPE\n        self.exchange = self._create_exchange(\n            exchange, exchange_type, self.delivery_mode,\n        )\n        self.serializer = serializer or conf.CELERY_RESULT_SERIALIZER\n        self.auto_delete = auto_delete\n        self.queue_arguments = dictfilter({\n            'x-expires': maybe_s_to_ms(self.expires),\n        })\n\n    def _create_exchange(self, name, type='direct', delivery_mode=2):\n        return self.Exchange(name=name,\n                             type=type,\n                             delivery_mode=delivery_mode,\n                             durable=self.persistent,\n                             auto_delete=False)\n\n    def _create_binding(self, task_id):\n        name = self.rkey(task_id)\n        return self.Queue(name=name,\n                          exchange=self.exchange,\n                          routing_key=name,\n                          durable=self.persistent,\n                          auto_delete=self.auto_delete,\n                          queue_arguments=self.queue_arguments)\n\n    def revive(self, channel):\n        pass\n\n    def rkey(self, task_id):\n        return task_id.replace('-', '')\n\n    def destination_for(self, task_id, request):\n        if request:\n            return self.rkey(task_id), request.correlation_id or task_id\n        return self.rkey(task_id), task_id\n\n    def store_result(self, task_id, result, status,\n                     traceback=None, request=None, **kwargs):\n        \"\"\"Send task return value and status.\"\"\"\n        routing_key, correlation_id = self.destination_for(task_id, request)\n        if not routing_key:\n            return\n        with self.app.amqp.producer_pool.acquire(block=True) as producer:\n            producer.publish(\n                {'task_id': task_id, 'status': status,\n                 'result': self.encode_result(result, status),\n                 'traceback': traceback,\n                 'children': self.current_task_children(request)},\n                exchange=self.exchange,\n                routing_key=routing_key,\n                correlation_id=correlation_id,\n                serializer=self.serializer,\n                retry=True, retry_policy=self.retry_policy,\n                declare=self.on_reply_declare(task_id),\n                delivery_mode=self.delivery_mode,\n            )\n        return result\n\n    def on_reply_declare(self, task_id):\n        return [self._create_binding(task_id)]\n\n    def wait_for(self, task_id, timeout=None, cache=True,\n                 no_ack=True, on_interval=None,\n                 READY_STATES=states.READY_STATES,\n                 PROPAGATE_STATES=states.PROPAGATE_STATES,\n                 **kwargs):\n        cached_meta = self._cache.get(task_id)\n        if cache and cached_meta and \\\n                cached_meta['status'] in READY_STATES:\n            return cached_meta\n        else:\n            try:\n                return self.consume(task_id, timeout=timeout, no_ack=no_ack,\n                                    on_interval=on_interval)\n            except socket.timeout:\n                raise TimeoutError('The operation timed out.')\n\n    def get_task_meta(self, task_id, backlog_limit=1000):\n        # Polling and using basic_get\n        with self.app.pool.acquire_channel(block=True) as (_, channel):\n            binding = self._create_binding(task_id)(channel)\n            binding.declare()\n\n            prev = latest = acc = None\n            for i in range(backlog_limit):  # spool ffwd\n                acc = binding.get(\n                    accept=self.accept, no_ack=False,\n                )\n                if not acc:  # no more messages\n                    break\n                if acc.payload['task_id'] == task_id:\n                    prev, latest = latest, acc\n                if prev:\n                    # backends are not expected to keep history,\n                    # so we delete everything except the most recent state.\n                    prev.ack()\n                    prev = None\n            else:\n                raise self.BacklogLimitExceeded(task_id)\n\n            if latest:\n                payload = self._cache[task_id] = latest.payload\n                latest.requeue()\n                return payload\n            else:\n                # no new state, use previous\n                try:\n                    return self._cache[task_id]\n                except KeyError:\n                    # result probably pending.\n                    return {'status': states.PENDING, 'result': None}\n    poll = get_task_meta  # XXX compat\n\n    def drain_events(self, connection, consumer,\n                     timeout=None, on_interval=None, now=monotonic, wait=None):\n        wait = wait or connection.drain_events\n        results = {}\n\n        def callback(meta, message):\n            if meta['status'] in states.READY_STATES:\n                results[meta['task_id']] = meta\n\n        consumer.callbacks[:] = [callback]\n        time_start = now()\n\n        while 1:\n            # Total time spent may exceed a single call to wait()\n            if timeout and now() - time_start >= timeout:\n                raise socket.timeout()\n            try:\n                wait(timeout=1)\n            except socket.timeout:\n                pass\n            if on_interval:\n                on_interval()\n            if results:  # got event on the wanted channel.\n                break\n        self._cache.update(results)\n        return results\n\n    def consume(self, task_id, timeout=None, no_ack=True, on_interval=None):\n        wait = self.drain_events\n        with self.app.pool.acquire_channel(block=True) as (conn, channel):\n            binding = self._create_binding(task_id)\n            with self.Consumer(channel, binding,\n                               no_ack=no_ack, accept=self.accept) as consumer:\n                while 1:\n                    try:\n                        return wait(\n                            conn, consumer, timeout, on_interval)[task_id]\n                    except KeyError:\n                        continue\n\n    def _many_bindings(self, ids):\n        return [self._create_binding(task_id) for task_id in ids]\n\n    def get_many(self, task_ids, timeout=None, no_ack=True, on_message=None,\n                 now=monotonic, getfields=itemgetter('status', 'task_id'),\n                 READY_STATES=states.READY_STATES,\n                 PROPAGATE_STATES=states.PROPAGATE_STATES, **kwargs):\n        with self.app.pool.acquire_channel(block=True) as (conn, channel):\n            ids = set(task_ids)\n            cached_ids = set()\n            mark_cached = cached_ids.add\n            for task_id in ids:\n                try:\n                    cached = self._cache[task_id]\n                except KeyError:\n                    pass\n                else:\n                    if cached['status'] in READY_STATES:\n                        yield task_id, cached\n                        mark_cached(task_id)\n            ids.difference_update(cached_ids)\n            results = deque()\n            push_result = results.append\n            push_cache = self._cache.__setitem__\n            decode_result = self.meta_from_decoded\n\n            def _on_message(message):\n                body = decode_result(message.decode())\n                if on_message is not None:\n                    on_message(body)\n                state, uid = getfields(body)\n                if state in READY_STATES:\n                    push_result(body) \\\n                        if uid in task_ids else push_cache(uid, body)\n\n            bindings = self._many_bindings(task_ids)\n            with self.Consumer(channel, bindings, on_message=_on_message,\n                               accept=self.accept, no_ack=no_ack):\n                wait = conn.drain_events\n                popleft = results.popleft\n                while ids:\n                    wait(timeout=timeout)\n                    while results:\n                        state = popleft()\n                        task_id = state['task_id']\n                        ids.discard(task_id)\n                        push_cache(task_id, state)\n                        yield task_id, state\n\n    def reload_task_result(self, task_id):\n        raise NotImplementedError(\n            'reload_task_result is not supported by this backend.')\n\n    def reload_group_result(self, task_id):\n        \"\"\"Reload group result, even if it has been previously fetched.\"\"\"\n        raise NotImplementedError(\n            'reload_group_result is not supported by this backend.')\n\n    def save_group(self, group_id, result):\n        raise NotImplementedError(\n            'save_group is not supported by this backend.')\n\n    def restore_group(self, group_id, cache=True):\n        raise NotImplementedError(\n            'restore_group is not supported by this backend.')\n\n    def delete_group(self, group_id):\n        raise NotImplementedError(\n            'delete_group is not supported by this backend.')\n\n    def __reduce__(self, args=(), kwargs={}):\n        kwargs.update(\n            connection=self._connection,\n            exchange=self.exchange.name,\n            exchange_type=self.exchange.type,\n            persistent=self.persistent,\n            serializer=self.serializer,\n            auto_delete=self.auto_delete,\n            expires=self.expires,\n        )\n        return super(AMQPBackend, self).__reduce__(args, kwargs)\n", "path": "celery/backends/amqp.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\"\"\"\n    celery.backends.amqp\n    ~~~~~~~~~~~~~~~~~~~~\n\n    The AMQP result backend.\n\n    This backend publishes results as messages.\n\n\"\"\"\nfrom __future__ import absolute_import\n\nimport socket\n\nfrom collections import deque\nfrom operator import itemgetter\n\nfrom kombu import Exchange, Queue, Producer, Consumer\n\nfrom celery import states\nfrom celery.exceptions import TimeoutError\nfrom celery.five import range, monotonic\nfrom celery.utils.functional import dictfilter\nfrom celery.utils.log import get_logger\nfrom celery.utils.timeutils import maybe_s_to_ms\n\nfrom .base import BaseBackend\n\n__all__ = ['BacklogLimitExceeded', 'AMQPBackend']\n\nlogger = get_logger(__name__)\n\n\nclass BacklogLimitExceeded(Exception):\n    \"\"\"Too much state history to fast-forward.\"\"\"\n\n\ndef repair_uuid(s):\n    # Historically the dashes in UUIDS are removed from AMQ entity names,\n    # but there is no known reason to.  Hopefully we'll be able to fix\n    # this in v4.0.\n    return '%s-%s-%s-%s-%s' % (s[:8], s[8:12], s[12:16], s[16:20], s[20:])\n\n\nclass NoCacheQueue(Queue):\n    can_cache_declaration = False\n\n\nclass AMQPBackend(BaseBackend):\n    \"\"\"Publishes results by sending messages.\"\"\"\n    Exchange = Exchange\n    Queue = NoCacheQueue\n    Consumer = Consumer\n    Producer = Producer\n\n    BacklogLimitExceeded = BacklogLimitExceeded\n\n    persistent = True\n    supports_autoexpire = True\n    supports_native_join = True\n\n    retry_policy = {\n        'max_retries': 20,\n        'interval_start': 0,\n        'interval_step': 1,\n        'interval_max': 1,\n    }\n\n    def __init__(self, app, connection=None, exchange=None, exchange_type=None,\n                 persistent=None, serializer=None, auto_delete=True, **kwargs):\n        super(AMQPBackend, self).__init__(app, **kwargs)\n        conf = self.app.conf\n        self._connection = connection\n        self.persistent = self.prepare_persistent(persistent)\n        self.delivery_mode = 2 if self.persistent else 1\n        exchange = exchange or conf.CELERY_RESULT_EXCHANGE\n        exchange_type = exchange_type or conf.CELERY_RESULT_EXCHANGE_TYPE\n        self.exchange = self._create_exchange(\n            exchange, exchange_type, self.delivery_mode,\n        )\n        self.serializer = serializer or conf.CELERY_RESULT_SERIALIZER\n        self.auto_delete = auto_delete\n        self.queue_arguments = dictfilter({\n            'x-expires': maybe_s_to_ms(self.expires),\n        })\n\n    def _create_exchange(self, name, type='direct', delivery_mode=2):\n        return self.Exchange(name=name,\n                             type=type,\n                             delivery_mode=delivery_mode,\n                             durable=self.persistent,\n                             auto_delete=False)\n\n    def _create_binding(self, task_id):\n        name = self.rkey(task_id)\n        return self.Queue(name=name,\n                          exchange=self.exchange,\n                          routing_key=name,\n                          durable=self.persistent,\n                          auto_delete=self.auto_delete,\n                          queue_arguments=self.queue_arguments)\n\n    def revive(self, channel):\n        pass\n\n    def rkey(self, task_id):\n        return task_id.replace('-', '')\n\n    def destination_for(self, task_id, request):\n        if request:\n            return self.rkey(task_id), request.correlation_id or task_id\n        return self.rkey(task_id), task_id\n\n    def store_result(self, task_id, result, status,\n                     traceback=None, request=None, **kwargs):\n        \"\"\"Send task return value and status.\"\"\"\n        routing_key, correlation_id = self.destination_for(task_id, request)\n        if not routing_key:\n            return\n        with self.app.amqp.producer_pool.acquire(block=True) as producer:\n            producer.publish(\n                {'task_id': task_id, 'status': status,\n                 'result': self.encode_result(result, status),\n                 'traceback': traceback,\n                 'children': self.current_task_children(request)},\n                exchange=self.exchange,\n                routing_key=routing_key,\n                correlation_id=correlation_id,\n                serializer=self.serializer,\n                retry=True, retry_policy=self.retry_policy,\n                declare=self.on_reply_declare(task_id),\n                delivery_mode=self.delivery_mode,\n            )\n        return result\n\n    def on_reply_declare(self, task_id):\n        return [self._create_binding(task_id)]\n\n    def wait_for(self, task_id, timeout=None, cache=True,\n                 no_ack=True, on_interval=None,\n                 READY_STATES=states.READY_STATES,\n                 PROPAGATE_STATES=states.PROPAGATE_STATES,\n                 **kwargs):\n        cached_meta = self._cache.get(task_id)\n        if cache and cached_meta and \\\n                cached_meta['status'] in READY_STATES:\n            return cached_meta\n        else:\n            try:\n                return self.consume(task_id, timeout=timeout, no_ack=no_ack,\n                                    on_interval=on_interval)\n            except socket.timeout:\n                raise TimeoutError('The operation timed out.')\n\n    def get_task_meta(self, task_id, backlog_limit=1000):\n        # Polling and using basic_get\n        with self.app.pool.acquire_channel(block=True) as (_, channel):\n            binding = self._create_binding(task_id)(channel)\n            binding.declare()\n\n            prev = latest = acc = None\n            for i in range(backlog_limit):  # spool ffwd\n                acc = binding.get(\n                    accept=self.accept, no_ack=False,\n                )\n                if not acc:  # no more messages\n                    break\n                if acc.payload['task_id'] == task_id:\n                    prev, latest = latest, acc\n                if prev:\n                    # backends are not expected to keep history,\n                    # so we delete everything except the most recent state.\n                    prev.ack()\n                    prev = None\n            else:\n                raise self.BacklogLimitExceeded(task_id)\n\n            if latest:\n                payload = self._cache[task_id] = latest.payload\n                latest.requeue()\n                return payload\n            else:\n                # no new state, use previous\n                try:\n                    return self._cache[task_id]\n                except KeyError:\n                    # result probably pending.\n                    return {'status': states.PENDING, 'result': None}\n    poll = get_task_meta  # XXX compat\n\n    def drain_events(self, connection, consumer,\n                     timeout=None, on_interval=None, now=monotonic, wait=None):\n        wait = wait or connection.drain_events\n        results = {}\n\n        def callback(meta, message):\n            if meta['status'] in states.READY_STATES:\n                results[meta['task_id']] = self.meta_from_decoded(meta)\n\n        consumer.callbacks[:] = [callback]\n        time_start = now()\n\n        while 1:\n            # Total time spent may exceed a single call to wait()\n            if timeout and now() - time_start >= timeout:\n                raise socket.timeout()\n            try:\n                wait(timeout=1)\n            except socket.timeout:\n                pass\n            if on_interval:\n                on_interval()\n            if results:  # got event on the wanted channel.\n                break\n        self._cache.update(results)\n        return results\n\n    def consume(self, task_id, timeout=None, no_ack=True, on_interval=None):\n        wait = self.drain_events\n        with self.app.pool.acquire_channel(block=True) as (conn, channel):\n            binding = self._create_binding(task_id)\n            with self.Consumer(channel, binding,\n                               no_ack=no_ack, accept=self.accept) as consumer:\n                while 1:\n                    try:\n                        return wait(\n                            conn, consumer, timeout, on_interval)[task_id]\n                    except KeyError:\n                        continue\n\n    def _many_bindings(self, ids):\n        return [self._create_binding(task_id) for task_id in ids]\n\n    def get_many(self, task_ids, timeout=None, no_ack=True, on_message=None,\n                 now=monotonic, getfields=itemgetter('status', 'task_id'),\n                 READY_STATES=states.READY_STATES,\n                 PROPAGATE_STATES=states.PROPAGATE_STATES, **kwargs):\n        with self.app.pool.acquire_channel(block=True) as (conn, channel):\n            ids = set(task_ids)\n            cached_ids = set()\n            mark_cached = cached_ids.add\n            for task_id in ids:\n                try:\n                    cached = self._cache[task_id]\n                except KeyError:\n                    pass\n                else:\n                    if cached['status'] in READY_STATES:\n                        yield task_id, cached\n                        mark_cached(task_id)\n            ids.difference_update(cached_ids)\n            results = deque()\n            push_result = results.append\n            push_cache = self._cache.__setitem__\n            decode_result = self.meta_from_decoded\n\n            def _on_message(message):\n                body = decode_result(message.decode())\n                if on_message is not None:\n                    on_message(body)\n                state, uid = getfields(body)\n                if state in READY_STATES:\n                    push_result(body) \\\n                        if uid in task_ids else push_cache(uid, body)\n\n            bindings = self._many_bindings(task_ids)\n            with self.Consumer(channel, bindings, on_message=_on_message,\n                               accept=self.accept, no_ack=no_ack):\n                wait = conn.drain_events\n                popleft = results.popleft\n                while ids:\n                    wait(timeout=timeout)\n                    while results:\n                        state = popleft()\n                        task_id = state['task_id']\n                        ids.discard(task_id)\n                        push_cache(task_id, state)\n                        yield task_id, state\n\n    def reload_task_result(self, task_id):\n        raise NotImplementedError(\n            'reload_task_result is not supported by this backend.')\n\n    def reload_group_result(self, task_id):\n        \"\"\"Reload group result, even if it has been previously fetched.\"\"\"\n        raise NotImplementedError(\n            'reload_group_result is not supported by this backend.')\n\n    def save_group(self, group_id, result):\n        raise NotImplementedError(\n            'save_group is not supported by this backend.')\n\n    def restore_group(self, group_id, cache=True):\n        raise NotImplementedError(\n            'restore_group is not supported by this backend.')\n\n    def delete_group(self, group_id):\n        raise NotImplementedError(\n            'delete_group is not supported by this backend.')\n\n    def __reduce__(self, args=(), kwargs={}):\n        kwargs.update(\n            connection=self._connection,\n            exchange=self.exchange.name,\n            exchange_type=self.exchange.type,\n            persistent=self.persistent,\n            serializer=self.serializer,\n            auto_delete=self.auto_delete,\n            expires=self.expires,\n        )\n        return super(AMQPBackend, self).__reduce__(args, kwargs)\n", "path": "celery/backends/amqp.py"}]}