import base64
from amqp.adapter.data_message_factory import DataMessageFactory
from amqp.adapter.logging_utils import logging_error, logging_info
from amqp.adapter.serializer import map_as_string, parse_map_string
from amqp.adapter.trace_info_adapter import TraceInfoAdapter
from amqp.model.model import ServiceMessage
from amqp.model.service_message_type import ServiceMessageType
from amqp.model.snowflake_id import SnowflakeId
RS = "|"
SPLIT_RS = r"\|"
INVALID_MESSAGE = ServiceMessage() # default values mean invalid message
[docs]
class ServiceMessageFactory:
"""
build/serialize/deserialize CleverMicro Service Message class
"""
[docs]
def __init__(self, name: str):
self.process_name = name
[docs]
def to_string(self, message: ServiceMessage) -> str:
"""
Convert ServiceMessage to human-readable format
:param message: ServiceMessage object
:return: as string
"""
if message is not None:
return (
f"ServiceMessage{{id={message.id.as_string()}|type={self.format_message_type(message)}|"
f"recipient={message.recipient_name}|replyTo={message.reply_to}|"
f"traceInfo={message.trace_info}|body={base64.b64encode(message.message.encode()).decode() if message.payload_type & 0x80 else message.message}}}"
)
return "null"
[docs]
def of(
self, message_type: ServiceMessageType, payload: str, recipient_name: str
) -> ServiceMessage:
"""
Builds ServiceMessage object from relevant values.
:param message_type: enum Message type
:param payload: Message content, stored as-is
:param recipient_name: Recipient name (from the routing expression)
:return: ServiceMessage object
"""
_id = DataMessageFactory.get_instance(1).generate_snowflake_id()
return ServiceMessage(
id=_id,
payload_type=0,
message_type=message_type,
recipient_name=recipient_name,
reply_to=self.process_name,
trace_info=TraceInfoAdapter.create_produce_span(_id),
message=payload,
)
[docs]
def as_base64(
self, message_type: ServiceMessageType, payload: str, recipient_name: str
) -> ServiceMessage:
"""
Builds ServiceMessage object from relevant values, stores content Base64 encoded.
:param message_type: enum Message type
:param payload: Message content, stored as Base64 string
:param recipient_name: Recipient name (from the routing expression)
:return: ServiceMessage object
"""
_id = DataMessageFactory.get_instance(1).generate_snowflake_id()
return ServiceMessage(
id=_id,
payload_type=0x80,
message_type=message_type,
recipient_name=recipient_name,
reply_to=self.process_name,
trace_info=TraceInfoAdapter.create_produce_span(_id.as_string()),
message=payload,
)
[docs]
def from_bytes(self, input_bytes: bytes) -> ServiceMessage:
"""
Deserialize ServiceMessage from byte array serialized form.
:param input_bytes: serialized ServiceMessage
:return: ServiceMessage object represented by the input.
"""
logging_info(f"ServiceMessage >>> [{input_bytes.decode()}]")
input_str_array = input_bytes.decode().split(RS)
try:
i = 0
_id = input_str_array[i]
i += 1
payload_type = int(input_str_array[i], 16)
i += 1
message_type = ServiceMessageType(payload_type + 1 & 0x7F)
recipient_name = input_str_array[i] if i < len(input_str_array) else ""
i += 1
reply_to = input_str_array[i] if i < len(input_str_array) else ""
i += 1
trace_info = parse_map_string(input_str_array[i]) if i < len(input_str_array) else {}
i += 1
message = input_str_array[i] if i < len(input_str_array) else ""
return ServiceMessage(
id=SnowflakeId.from_hex(_id),
payload_type=payload_type,
message_type=message_type,
recipient_name=recipient_name,
reply_to=reply_to,
trace_info=trace_info,
message=(base64.b64decode(message).decode() if payload_type & 0x80 else message),
)
except Exception as e:
logging_error(
"Unexpected ServiceMessage format. Expected 6 tokens separated by '|' in format: "
"[ID(64bit number)|MessageType(8bit number)|RecipientName(String)|"
"ReplyTo(String)|TraceInfo|Payload(String or Base64)], got: "
"[%s], reported problem: %s",
input_bytes.decode(),
str(e),
)
return INVALID_MESSAGE
[docs]
def serialize(self, message: ServiceMessage) -> bytes:
"""
Serialize ServiceMessage to byte array.
:param message: ServiceMessage to serialize.
:return: serialized object.
"""
return (
f"{message.id.as_string()}{RS}{self.format_message_type(message)}{RS}{message.recipient_name}{RS}"
f"{message.reply_to}{RS}{map_as_string(message.trace_info)}{RS}"
f"{base64.b64encode(message.message.encode()).decode() if message.payload_type != 0 else message.message}"
).encode("utf-8")