gzip message data that exceeds 60000 bytes (mainly to give more flexibility around some default SNS message limits)
import ujson
import uuid
import time
+import zlib
+import base64
from typing import Any, Dict, Tuple, Union
PROTOCOL_VERSION = 'json_base-wip'
@@ -11,6 +13,12 @@ class JsonBase(object):
@classmethod
async def build_message(cls, service: Any, topic: str, data: Any) -> str:
_uuid = str(uuid.uuid4())
+
+ data_encoding = 'raw'
+ if len(ujson.dumps(data)) >= 60000 or True:
+ data = base64.b64encode(zlib.compress(ujson.dumps(data).encode('utf-8'))).decode('utf-8')
+ data_encoding = 'base64_gzip_json'
+
message = {
'service': {
'name': getattr(service, 'name', None),
@@ -21,7 +29,8 @@ class JsonBase(object):
'protocol_version': PROTOCOL_VERSION,
'compatible_protocol_versions': COMPATIBLE_PROTOCOL_VERSIONS,
'timestamp': time.time(),
- 'topic': topic
+ 'topic': topic,
+ 'data_encoding': data_encoding
},
'data': data
}
@@ -37,6 +46,10 @@ class JsonBase(object):
if PROTOCOL_VERSION not in compatible_protocol_versions:
return False, message_uuid, timestamp
+ data = message.get('data')
+ if message.get('metadata', {}).get('data_encoding') == 'base64_gzip_json':
+ data = ujson.loads(zlib.decompress(base64.b64decode(message.get('data').encode('utf-8'))).decode('utf-8'))
+
return {
'service': {
'name': message.get('service', {}).get('name'),
@@ -47,7 +60,8 @@ class JsonBase(object):
'protocol_version': message.get('metadata', {}).get('protocol_version'),
'compatible_protocol_versions': message.get('metadata', {}).get('compatible_protocol_versions'),
'timestamp': message.get('metadata', {}).get('timestamp'),
- 'topic': message.get('metadata', {}).get('topic')
+ 'topic': message.get('metadata', {}).get('topic'),
+ 'data_encoding': message.get('metadata', {}).get('data_encoding')
},
- 'data': message.get('data')
+ 'data': data
}, message_uuid, timestamp
