From 05ede8f86c71ae21a331961b7795c0bd21ef09df Mon Sep 17 00:00:00 2001 From: tierno Date: Mon, 28 Jan 2019 16:20:18 +0000 Subject: [PATCH] msgkafka, provide loop on config Change-Id: I974ce861c6ff88e61594c9983faff5459ac5708a Signed-off-by: tierno --- osm_common/__init__.py | 4 ++-- osm_common/msgkafka.py | 14 +++++++++++--- osm_common/msglocal.py | 6 +++++- 3 files changed, 18 insertions(+), 6 deletions(-) diff --git a/osm_common/__init__.py b/osm_common/__init__.py index 73fd77e..eb858ee 100644 --- a/osm_common/__init__.py +++ b/osm_common/__init__.py @@ -15,6 +15,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -version = '0.1.15' +version = '0.1.16' # TODO add package version filling commit id with 0's; e.g.: '5.0.0.post11+00000000.dirty-1' -date_version = '2018-12-05' +date_version = '2019-01-28' diff --git a/osm_common/msgkafka.py b/osm_common/msgkafka.py index b782685..63ee69c 100644 --- a/osm_common/msgkafka.py +++ b/osm_common/msgkafka.py @@ -42,7 +42,7 @@ class MsgKafka(MsgBase): self.logger = logging.getLogger(config["logger_name"]) self.host = config["host"] self.port = config["port"] - self.loop = asyncio.get_event_loop() + self.loop = config.get("loop") or asyncio.get_event_loop() self.broker = str(self.host) + ":" + str(self.port) self.group_id = config.get("group_id") @@ -84,6 +84,14 @@ class MsgKafka(MsgBase): raise MsgException("Error reading {} topic: {}".format(topic, str(e))) async def aiowrite(self, topic, key, msg, loop=None): + """ + Asyncio write + :param topic: str kafka topic + :param key: str kafka key + :param msg: str or dictionary kafka message + :param loop: asyncio loop. To be DEPRECATED! in near future!!! loop must be provided inside config at connect + :return: None + """ if not loop: loop = self.loop @@ -99,9 +107,9 @@ class MsgKafka(MsgBase): async def aioread(self, topic, loop=None, callback=None, aiocallback=None, **kwargs): """ - Asyncio read from one or several topics. It blocks. + Asyncio read from one or several topics. :param topic: can be str: single topic; or str list: several topics - :param loop: asyncio loop + :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 in kafka bus :param aiocallback: async callback function that will handle the message in kafka bus :param kwargs: optional keyword arguments for callback function diff --git a/osm_common/msglocal.py b/osm_common/msglocal.py index 1e8e089..eba4b97 100644 --- a/osm_common/msglocal.py +++ b/osm_common/msglocal.py @@ -42,6 +42,7 @@ class MsgLocal(MsgBase): self.files_read = {} self.files_write = {} self.buffer = {} + self.loop = None def connect(self, config): try: @@ -52,6 +53,8 @@ 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 @@ -127,6 +130,7 @@ class MsgLocal(MsgBase): :param loop: asyncio loop :return: topic, key, message """ + _loop = loop or self.loop try: while True: msg = self.read(topic, blocks=False) @@ -137,7 +141,7 @@ class MsgLocal(MsgBase): await aiocallback(*msg, **kwargs) else: return msg - await asyncio.sleep(2, loop=loop) + await asyncio.sleep(2, loop=_loop) except MsgException: raise except Exception as e: # TODO refine -- 2.17.1