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.
19 from http
import HTTPStatus
22 from time
import sleep
24 from osm_common
.msgbase
import MsgBase
, MsgException
27 __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 as read_topic_error
:
68 if isinstance(read_topic_error
, (IOError, FileNotFoundError
)):
69 self
.logger
.exception(
70 f
"{read_topic_error} occured while closing read topic files."
72 elif isinstance(read_topic_error
, KeyError):
73 self
.logger
.exception(
74 f
"{read_topic_error} occured while reading from files_read dictionary."
77 self
.logger
.exception(
78 f
"{read_topic_error} occured while closing read topics."
81 for topic
, f
in self
.files_write
.items():
84 self
.files_write
[topic
] = None
85 except Exception as write_topic_error
:
86 if isinstance(write_topic_error
, (IOError, FileNotFoundError
)):
87 self
.logger
.exception(
88 f
"{write_topic_error} occured while closing write topic files."
90 elif isinstance(write_topic_error
, KeyError):
91 self
.logger
.exception(
92 f
"{write_topic_error} occured while reading from files_write dictionary."
95 self
.logger
.exception(
96 f
"{write_topic_error} occured while closing write topics."
99 def write(self
, topic
, key
, msg
):
101 Insert a message into topic
103 :param key: key text to be inserted
104 :param msg: value object to be inserted, can be str, object ...
105 :return: None or raises and exception
109 if topic
not in self
.files_write
:
110 self
.files_write
[topic
] = open(self
.path
+ topic
, "a+")
113 self
.files_write
[topic
],
114 default_flow_style
=True,
117 self
.files_write
[topic
].flush()
118 except Exception as e
: # TODO refine
119 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
121 def read(self
, topic
, blocks
=True):
123 Read from one or several topics. it is non blocking returning None if nothing is available
124 :param topic: can be str: single topic; or str list: several topics
125 :param blocks: indicates if it should wait and block until a message is present or returns None
126 :return: topic, key, message; or None if blocks==True
129 if isinstance(topic
, (list, tuple)):
132 topic_list
= (topic
,)
134 for single_topic
in topic_list
:
136 if single_topic
not in self
.files_read
:
137 self
.files_read
[single_topic
] = open(
138 self
.path
+ single_topic
, "a+"
140 self
.buffer[single_topic
] = ""
141 self
.buffer[single_topic
] += self
.files_read
[
144 if not self
.buffer[single_topic
].endswith("\n"):
146 msg_dict
= yaml
.safe_load(self
.buffer[single_topic
])
147 self
.buffer[single_topic
] = ""
148 if len(msg_dict
) != 1:
150 "Length of message dictionary is not equal to 1"
152 for k
, v
in msg_dict
.items():
153 return single_topic
, k
, v
157 except Exception as e
: # TODO refine
158 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
161 self
, topic
, loop
=None, callback
=None, aiocallback
=None, group_id
=None, **kwargs
164 Asyncio read from one or several topics. It blocks
165 :param topic: can be str: single topic; or str list: several topics
166 :param loop: asyncio loop. To be DEPRECATED! in near future!!! loop must be provided inside config at connect
167 :param callback: synchronous callback function that will handle the message
168 :param aiocallback: async callback function that will handle the message
169 :param group_id: group_id to use for load balancing. Can be False (set group_id to None), None (use general
170 group_id provided at connect inside config), or a group_id string
171 :param kwargs: optional keyword arguments for callback function
172 :return: If no callback defined, it returns (topic, key, message)
174 _loop
= loop
or self
.loop
177 msg
= self
.read(topic
, blocks
=False)
180 callback(*msg
, **kwargs
)
182 await aiocallback(*msg
, **kwargs
)
185 await asyncio
.sleep(2, loop
=_loop
)
188 except Exception as e
: # TODO refine
189 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
191 async def aiowrite(self
, topic
, key
, msg
, loop
=None):
193 Asyncio write. It blocks
196 :param msg: message, can be str or yaml
197 :param loop: asyncio loop
198 :return: nothing if ok or raises an exception
200 return self
.write(topic
, key
, msg
)