Tests passing

celery/backends/amqp.py

@@ -256,10 +256,10 @@ class AMQPBackend(BaseBackend):
             results = deque()
             push_result = results.append
             push_cache = self._cache.__setitem__
-            decode_result = self.decode_result
+            decode_result = self.meta_from_decoded
 
             def on_message(message):
-                body = decode_result(message.body)
+                body = decode_result(message.decode())
                 state, uid = getfields(body)
                 if state in READY_STATES:
                     push_result(body) \

celery/backends/base.py

@@ -181,12 +181,14 @@ class BaseBackend(object):
         _, _, payload = dumps(data, serializer=self.serializer)
         return payload
 
-    def decode_result(self, payload):
-        meta = self.decode(payload)
+    def meta_from_decoded(self, meta):
         if meta['status'] in self.EXCEPTION_STATES:
             meta['result'] = self.exception_to_python(meta['result'])
         return meta
 
+    def decode_result(self, payload):
+        return self.meta_from_decoded(self.decode(payload))
+
     def decode(self, payload):
         payload = PY3 and payload or str(payload)
         return loads(payload,