X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;ds=sidebyside;f=osm_common%2Fmsglocal.py;h=6d4cb58fed1200c4ea78fdf640358be181d62a55;hb=c04a88c04823353dedf4145c6ae359ad041d111c;hp=f731e741fdd801264f86c3c4b97fd630f48c4d88;hpb=87858cab98b3b169fc891fd2e0a0ba10f8b46127;p=osm%2Fcommon.git diff --git a/osm_common/msglocal.py b/osm_common/msglocal.py index f731e74..6d4cb58 100644 --- a/osm_common/msglocal.py +++ b/osm_common/msglocal.py @@ -15,15 +15,16 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio +from http import HTTPStatus import logging import os -import yaml -import asyncio -from osm_common.msgbase import MsgBase, MsgException from time import sleep -__author__ = "Alfonso Tierno " +from osm_common.msgbase import MsgBase, MsgException +import yaml +__author__ = "Alfonso Tierno " """ This emulated kafka bus by just using a shared file system. Useful for testing or devops. One file is used per topic. Only one producer and one consumer is allowed per topic. Both consumer and producer @@ -33,14 +34,14 @@ One text line per message is used in yaml format. class MsgLocal(MsgBase): - - def __init__(self, logger_name='msg'): - self.logger = logging.getLogger(logger_name) + def __init__(self, logger_name="msg", lock=False): + super().__init__(logger_name, lock) self.path = None # create a different file for each topic self.files_read = {} self.files_write = {} self.buffer = {} + self.loop = None def connect(self, config): try: @@ -51,22 +52,49 @@ class MsgLocal(MsgBase): self.path += "/" if not os.path.exists(self.path): os.mkdir(self.path) + self.loop = config.get("loop") + except MsgException: raise except Exception as e: # TODO refine - raise MsgException(str(e)) + raise MsgException(str(e), http_code=HTTPStatus.INTERNAL_SERVER_ERROR) def disconnect(self): - for f in self.files_read.values(): + for topic, f in self.files_read.items(): try: f.close() - except Exception: # TODO refine - pass - for f in self.files_write.values(): + self.files_read[topic] = None + except Exception as read_topic_error: + if isinstance(read_topic_error, (IOError, FileNotFoundError)): + self.logger.exception( + f"{read_topic_error} occured while closing read topic files." + ) + elif isinstance(read_topic_error, KeyError): + self.logger.exception( + f"{read_topic_error} occured while reading from files_read dictionary." + ) + else: + self.logger.exception( + f"{read_topic_error} occured while closing read topics." + ) + + for topic, f in self.files_write.items(): try: f.close() - except Exception: # TODO refine - pass + self.files_write[topic] = None + except Exception as write_topic_error: + if isinstance(write_topic_error, (IOError, FileNotFoundError)): + self.logger.exception( + f"{write_topic_error} occured while closing write topic files." + ) + elif isinstance(write_topic_error, KeyError): + self.logger.exception( + f"{write_topic_error} occured while reading from files_write dictionary." + ) + else: + self.logger.exception( + f"{write_topic_error} occured while closing write topics." + ) def write(self, topic, key, msg): """ @@ -77,12 +105,18 @@ class MsgLocal(MsgBase): :return: None or raises and exception """ try: - if topic not in self.files_write: - self.files_write[topic] = open(self.path + topic, "a+") - yaml.safe_dump({key: msg}, self.files_write[topic], default_flow_style=True, width=20000) - self.files_write[topic].flush() + with self.lock: + if topic not in self.files_write: + self.files_write[topic] = open(self.path + topic, "a+") + yaml.safe_dump( + {key: msg}, + self.files_write[topic], + default_flow_style=True, + width=20000, + ) + self.files_write[topic].flush() except Exception as e: # TODO refine - raise MsgException(str(e)) + raise MsgException(str(e), HTTPStatus.INTERNAL_SERVER_ERROR) def read(self, topic, blocks=True): """ @@ -95,43 +129,64 @@ class MsgLocal(MsgBase): if isinstance(topic, (list, tuple)): topic_list = topic else: - topic_list = (topic, ) + topic_list = (topic,) while True: for single_topic in topic_list: - if single_topic not in self.files_read: - self.files_read[single_topic] = open(self.path + single_topic, "a+") + with self.lock: + if single_topic not in self.files_read: + self.files_read[single_topic] = open( + self.path + single_topic, "a+" + ) + self.buffer[single_topic] = "" + self.buffer[single_topic] += self.files_read[ + single_topic + ].readline() + if not self.buffer[single_topic].endswith("\n"): + continue + msg_dict = yaml.safe_load(self.buffer[single_topic]) self.buffer[single_topic] = "" - self.buffer[single_topic] += self.files_read[single_topic].readline() - if not self.buffer[single_topic].endswith("\n"): - continue - msg_dict = yaml.load(self.buffer[single_topic]) - self.buffer[single_topic] = "" - assert len(msg_dict) == 1 - for k, v in msg_dict.items(): - return single_topic, k, v + if len(msg_dict) != 1: + raise ValueError( + "Length of message dictionary is not equal to 1" + ) + for k, v in msg_dict.items(): + return single_topic, k, v if not blocks: return None sleep(2) except Exception as e: # TODO refine - raise MsgException(str(e)) + raise MsgException(str(e), HTTPStatus.INTERNAL_SERVER_ERROR) - async def aioread(self, topic, loop): + async def aioread( + self, topic, loop=None, callback=None, aiocallback=None, group_id=None, **kwargs + ): """ Asyncio read from one or several topics. It blocks :param topic: can be str: single topic; or str list: several topics - :param loop: asyncio loop - :return: topic, key, message + :param loop: asyncio loop. To be DEPRECATED! in near future!!! loop must be provided inside config at connect + :param callback: synchronous callback function that will handle the message + :param aiocallback: async callback function that will handle the message + :param group_id: group_id to use for load balancing. Can be False (set group_id to None), None (use general + group_id provided at connect inside config), or a group_id string + :param kwargs: optional keyword arguments for callback function + :return: If no callback defined, it returns (topic, key, message) """ + _loop = loop or self.loop try: while True: msg = self.read(topic, blocks=False) if msg: - return msg - await asyncio.sleep(2, loop=loop) + if callback: + callback(*msg, **kwargs) + elif aiocallback: + await aiocallback(*msg, **kwargs) + else: + return msg + await asyncio.sleep(2, loop=_loop) except MsgException: raise except Exception as e: # TODO refine - raise MsgException(str(e)) + raise MsgException(str(e), HTTPStatus.INTERNAL_SERVER_ERROR) async def aiowrite(self, topic, key, msg, loop=None): """