X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_common%2Fmsglocal.py;h=c10ff17b0808c6e649755f3e7886b33db8d4ecf3;hb=2644b76248a1b96f7a47013b414e31b4e3feecf8;hp=843b3766a9dacf475294f11757b094282a681ea9;hpb=6472e2b6e90c263b59c1195a952fbf4b7d35eba0;p=osm%2Fcommon.git diff --git a/osm_common/msglocal.py b/osm_common/msglocal.py index 843b376..c10ff17 100644 --- a/osm_common/msglocal.py +++ b/osm_common/msglocal.py @@ -34,8 +34,7 @@ One text line per message is used in yaml format. class MsgLocal(MsgBase): - - def __init__(self, logger_name='msg', lock=False): + def __init__(self, logger_name="msg", lock=False): super().__init__(logger_name, lock) self.path = None # create a different file for each topic @@ -86,7 +85,12 @@ class MsgLocal(MsgBase): 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) + 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), HTTPStatus.INTERNAL_SERVER_ERROR) @@ -102,14 +106,18 @@ 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: with self.lock: if single_topic not in self.files_read: - self.files_read[single_topic] = open(self.path + single_topic, "a+") + 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() + 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]) @@ -123,7 +131,9 @@ class MsgLocal(MsgBase): except Exception as e: # TODO refine raise MsgException(str(e), HTTPStatus.INTERNAL_SERVER_ERROR) - async def aioread(self, topic, loop=None, callback=None, aiocallback=None, group_id=None, **kwargs): + 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