osm_nbi/msglocal.py

   1 import logging
   2 import os
   3 import yaml
   4 import asyncio
   5 from msgbase import MsgBase, MsgException
   6 from time import sleep
   7
   8 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
   9
  10 """
  11 This emulated kafka bus by just using a shared file system. Usefull for testing or devops.
  12 One file is used per topic. Only one producer and one consumer is allowed per topic. Both consumer and producer
  13 access to the same file. e.g. same volume if running with docker.
  14 One text line per message is used in yaml format
  15 """
  16
  17 class MsgLocal(MsgBase):
  18
  19     def __init__(self, logger_name='msg'):
  20         self.logger = logging.getLogger(logger_name)
  21         self.path = None
  22         # create a different file for each topic
  23         self.files = {}
  24         self.buffer = {}
  25
  26     def connect(self, config):
  27         try:
  28             if "logger_name" in config:
  29                 self.logger = logging.getLogger(config["logger_name"])
  30             self.path = config["path"]
  31             if not self.path.endswith("/"):
  32                 self.path += "/"
  33             if not os.path.exists(self.path):
  34                 os.mkdir(self.path)
  35         except MsgException:
  36             raise
  37         except Exception as e:  # TODO refine
  38             raise MsgException(str(e))
  39
  40     def disconnect(self):
  41         for f in self.files.values():
  42             try:
  43                 f.close()
  44             except Exception as e:  # TODO refine
  45                 pass
  46
  47     def write(self, topic, key, msg):
  48         """
  49         Insert a message into topic
  50         :param topic: topic
  51         :param key: key text to be inserted
  52         :param msg: value object to be inserted, can be str, object ...
  53         :return: None or raises and exception
  54         """
  55         try:
  56             if topic not in self.files:
  57                 self.files[topic] = open(self.path + topic, "a+")
  58             yaml.safe_dump({key: msg}, self.files[topic], default_flow_style=True, width=20000)
  59             self.files[topic].flush()
  60         except Exception as e:  # TODO refine
  61             raise MsgException(str(e))
  62
  63     def read(self, topic, blocks=True):
  64         """
  65         Read from one or several topics. it is non blocking returning None if nothing is available
  66         :param topic: can be str: single topic; or str list: several topics
  67         :param blocks: indicates if it should wait and block until a message is present or returns None
  68         :return: topic, key, message; or None if blocks==True
  69         """
  70         try:
  71             if isinstance(topic, (list, tuple)):
  72                 topic_list = topic
  73             else:
  74                 topic_list = (topic, )
  75             while True:
  76                 for single_topic in topic_list:
  77                     if single_topic not in self.files:
  78                         self.files[single_topic] = open(self.path + single_topic, "a+")
  79                         self.buffer[single_topic] = ""
  80                     self.buffer[single_topic] += self.files[single_topic].readline()
  81                     if not self.buffer[single_topic].endswith("\n"):
  82                         continue
  83                     msg_dict = yaml.load(self.buffer[single_topic])
  84                     self.buffer[single_topic] = ""
  85                     assert len(msg_dict) == 1
  86                     for k, v in msg_dict.items():
  87                         return single_topic, k, v
  88                 if not blocks:
  89                     return None
  90                 sleep(2)
  91         except Exception as e:  # TODO refine
  92             raise MsgException(str(e))
  93
  94     async def aioread(self, topic, loop):
  95         """
  96         Asyncio read from one or several topics. It blocks
  97         :param topic: can be str: single topic; or str list: several topics
  98         :param loop: asyncio loop
  99         :return: topic, key, message
 100         """
 101         try:
 102             while True:
 103                 msg = self.read(topic, blocks=False)
 104                 if msg:
 105                     return msg
 106                 await asyncio.sleep(2, loop=loop)
 107         except MsgException:
 108             raise
 109         except Exception as e:  # TODO refine
 110             raise MsgException(str(e))
 111