fixed msgpack codec

Author: voith

frontera/contrib/backends/remote/codecs/msgpack.py

@@ -6,31 +6,11 @@
 from msgpack import packb, unpackb
 
 from frontera.core.codec import BaseDecoder, BaseEncoder
-import six
 from w3lib.util import to_native_str
 
 
 def _prepare_request_message(request):
-    def serialize(obj):
-        """Recursively walk object's hierarchy."""
-        if isinstance(obj, six.text_type):
-            return obj.encode('utf8')
-        if isinstance(obj, (bool, six.integer_types, float, six.binary_type)):
-            return obj
-        elif isinstance(obj, dict):
-            obj = obj.copy()
-            for key in obj:
-                obj[key] = serialize(obj[key])
-            return obj
-        elif isinstance(obj, list):
-            return [serialize(item) for item in obj]
-        elif isinstance(obj, tuple):
-            return tuple(serialize([item for item in obj]))
-        elif hasattr(obj, '__dict__'):
-            return serialize(obj.__dict__)
-        else:
-            return None
-    return [request.url, request.method, request.headers, request.cookies, serialize(request.meta)]
+    return [request.url, request.method, request.headers, request.cookies, request.meta]
 
 
 def _prepare_response_message(response, send_body):
@@ -42,28 +22,29 @@ def __init__(self, request_model, *a, **kw):
         self.send_body = True if 'send_body' in kw and kw['send_body'] else False
 
     def encode_add_seeds(self, seeds):
-        return packb([b'as', [_prepare_request_message(seed) for seed in seeds]])
+        return packb([b'as', [_prepare_request_message(seed) for seed in seeds]], use_bin_type=True, encoding="utf-8")
 
     def encode_page_crawled(self, response):
-        return packb([b'pc', _prepare_response_message(response, self.send_body)])
+        return packb([b'pc', _prepare_response_message(response, self.send_body)], use_bin_type=True, encoding="utf-8")
 
     def encode_links_extracted(self, request, links):
-        return packb([b'le', _prepare_request_message(request), [_prepare_request_message(link) for link in links]])
+        return packb([b'le', _prepare_request_message(request), [_prepare_request_message(link) for link in links]],
+                     use_bin_type=True, encoding="utf-8")
 
     def encode_request_error(self, request, error):
-        return packb([b're', _prepare_request_message(request), str(error)])
+        return packb([b're', _prepare_request_message(request), str(error)], use_bin_type=True, encoding="utf-8")
 
     def encode_request(self, request):
-        return packb(_prepare_request_message(request))
+        return packb(_prepare_request_message(request), use_bin_type=True, encoding="utf-8")
 
     def encode_update_score(self, request, score, schedule):
-        return packb([b'us', _prepare_request_message(request), score, schedule])
+        return packb([b'us', _prepare_request_message(request), score, schedule], use_bin_type=True, encoding="utf-8")
 
     def encode_new_job_id(self, job_id):
-        return packb([b'njid', int(job_id)])
+        return packb([b'njid', int(job_id)], use_bin_type=True, encoding="utf-8")
 
     def encode_offset(self, partition_id, offset):
-        return packb([b'of', int(partition_id), int(offset)])
+        return packb([b'of', int(partition_id), int(offset)], use_bin_type=True, encoding="utf-8")
 
 
 class Decoder(BaseDecoder):
@@ -87,7 +68,7 @@ def _request_from_object(self, obj):
                                    meta=obj[4])
 
     def decode(self, buffer):
-        obj = unpackb(buffer)
+        obj = unpackb(buffer, encoding="utf-8")
         if obj[0] == b'pc':
             return ('page_crawled',
                     self._response_from_object(obj[1]))
@@ -108,4 +89,4 @@ def decode(self, buffer):
         return TypeError('Unknown message type')
 
     def decode_request(self, buffer):
-        return self._request_from_object(unpackb(buffer))
+        return self._request_from_object(unpackb(buffer, encoding="utf-8"))