1 # -*- coding: utf-8 -*-
3 # Copyright 2018 Telefonica S.A.
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
22 from osm_common
.msgbase
import MsgBase
, MsgException
23 from time
import sleep
24 from http
import HTTPStatus
26 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
29 This emulated kafka bus by just using a shared file system. Useful for testing or devops.
30 One file is used per topic. Only one producer and one consumer is allowed per topic. Both consumer and producer
31 access to the same file. e.g. same volume if running with docker.
32 One text line per message is used in yaml format.
36 class MsgLocal(MsgBase
):
37 def __init__(self
, logger_name
="msg", lock
=False):
38 super().__init
__(logger_name
, lock
)
40 # create a different file for each topic
46 def connect(self
, config
):
48 if "logger_name" in config
:
49 self
.logger
= logging
.getLogger(config
["logger_name"])
50 self
.path
= config
["path"]
51 if not self
.path
.endswith("/"):
53 if not os
.path
.exists(self
.path
):
55 self
.loop
= config
.get("loop")
59 except Exception as e
: # TODO refine
60 raise MsgException(str(e
), http_code
=HTTPStatus
.INTERNAL_SERVER_ERROR
)
63 for topic
, f
in self
.files_read
.items():
66 self
.files_read
[topic
] = None
67 except Exception: # TODO refine
69 for topic
, f
in self
.files_write
.items():
72 self
.files_write
[topic
] = None
73 except Exception: # TODO refine
76 def write(self
, topic
, key
, msg
):
78 Insert a message into topic
80 :param key: key text to be inserted
81 :param msg: value object to be inserted, can be str, object ...
82 :return: None or raises and exception
86 if topic
not in self
.files_write
:
87 self
.files_write
[topic
] = open(self
.path
+ topic
, "a+")
90 self
.files_write
[topic
],
91 default_flow_style
=True,
94 self
.files_write
[topic
].flush()
95 except Exception as e
: # TODO refine
96 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
98 def read(self
, topic
, blocks
=True):
100 Read from one or several topics. it is non blocking returning None if nothing is available
101 :param topic: can be str: single topic; or str list: several topics
102 :param blocks: indicates if it should wait and block until a message is present or returns None
103 :return: topic, key, message; or None if blocks==True
106 if isinstance(topic
, (list, tuple)):
109 topic_list
= (topic
,)
111 for single_topic
in topic_list
:
113 if single_topic
not in self
.files_read
:
114 self
.files_read
[single_topic
] = open(
115 self
.path
+ single_topic
, "a+"
117 self
.buffer[single_topic
] = ""
118 self
.buffer[single_topic
] += self
.files_read
[
121 if not self
.buffer[single_topic
].endswith("\n"):
123 msg_dict
= yaml
.safe_load(self
.buffer[single_topic
])
124 self
.buffer[single_topic
] = ""
125 assert len(msg_dict
) == 1
126 for k
, v
in msg_dict
.items():
127 return single_topic
, k
, v
131 except Exception as e
: # TODO refine
132 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
135 self
, topic
, loop
=None, callback
=None, aiocallback
=None, group_id
=None, **kwargs
138 Asyncio read from one or several topics. It blocks
139 :param topic: can be str: single topic; or str list: several topics
140 :param loop: asyncio loop. To be DEPRECATED! in near future!!! loop must be provided inside config at connect
141 :param callback: synchronous callback function that will handle the message
142 :param aiocallback: async callback function that will handle the message
143 :param group_id: group_id to use for load balancing. Can be False (set group_id to None), None (use general
144 group_id provided at connect inside config), or a group_id string
145 :param kwargs: optional keyword arguments for callback function
146 :return: If no callback defined, it returns (topic, key, message)
148 _loop
= loop
or self
.loop
151 msg
= self
.read(topic
, blocks
=False)
154 callback(*msg
, **kwargs
)
156 await aiocallback(*msg
, **kwargs
)
159 await asyncio
.sleep(2, loop
=_loop
)
162 except Exception as e
: # TODO refine
163 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
165 async def aiowrite(self
, topic
, key
, msg
, loop
=None):
167 Asyncio write. It blocks
170 :param msg: message, can be str or yaml
171 :param loop: asyncio loop
172 :return: nothing if ok or raises an exception
174 return self
.write(topic
, key
, msg
)