# Copyright (c) 2021, NVIDIA CORPORATION. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import uuid
[docs]class ReturnCode(object):
OK = "_ok"
ERROR = "_error"
[docs]class Message(object):
def __init__(self, topic: str, body):
"""To init a Message.
Args:
topic: message topic
body: message body.
"""
self.id = str(uuid.uuid4())
self.topic = topic
self.body = body
self.headers = {}
[docs] def get_ref_id(self, default=None):
return self.get_header(MsgHeader.REF_MSG_ID, default)
[docs] def set_ref_id(self, msg_id):
self.set_header(MsgHeader.REF_MSG_ID, msg_id)
[docs]def error_reply(err: str, meta: dict = None) -> Message:
msg = Message(topic="reply", body=err)
msg.set_header(MsgHeader.RETURN_CODE, ReturnCode.ERROR)
if meta:
msg.set_meta(meta)
return msg
[docs]def ok_reply(topic=None, body=None, meta: dict = None) -> Message:
if body is None:
body = "ok"
if topic is None:
topic = "reply"
msg = Message(topic=topic, body=body)
msg.set_header(MsgHeader.RETURN_CODE, ReturnCode.OK)
if meta:
msg.set_meta(meta)
return msg