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
45 def connect(self
, config
):
47 if "logger_name" in config
:
48 self
.logger
= logging
.getLogger(config
["logger_name"])
49 self
.path
= config
["path"]
50 if not self
.path
.endswith("/"):
52 if not os
.path
.exists(self
.path
):
57 except Exception as e
: # TODO refine
58 raise MsgException(str(e
), http_code
=HTTPStatus
.INTERNAL_SERVER_ERROR
)
61 for topic
, f
in self
.files_read
.items():
64 self
.files_read
[topic
] = None
65 except Exception as read_topic_error
:
66 if isinstance(read_topic_error
, (IOError, FileNotFoundError
)):
67 self
.logger
.exception(
68 f
"{read_topic_error} occured while closing read topic files."
70 elif isinstance(read_topic_error
, KeyError):
71 self
.logger
.exception(
72 f
"{read_topic_error} occured while reading from files_read dictionary."
75 self
.logger
.exception(
76 f
"{read_topic_error} occured while closing read topics."
79 for topic
, f
in self
.files_write
.items():
82 self
.files_write
[topic
] = None
83 except Exception as write_topic_error
:
84 if isinstance(write_topic_error
, (IOError, FileNotFoundError
)):
85 self
.logger
.exception(
86 f
"{write_topic_error} occured while closing write topic files."
88 elif isinstance(write_topic_error
, KeyError):
89 self
.logger
.exception(
90 f
"{write_topic_error} occured while reading from files_write dictionary."
93 self
.logger
.exception(
94 f
"{write_topic_error} occured while closing write topics."
97 def write(self
, topic
, key
, msg
):
99 Insert a message into topic
101 :param key: key text to be inserted
102 :param msg: value object to be inserted, can be str, object ...
103 :return: None or raises and exception
107 if topic
not in self
.files_write
:
108 self
.files_write
[topic
] = open(self
.path
+ topic
, "a+")
111 self
.files_write
[topic
],
112 default_flow_style
=True,
115 self
.files_write
[topic
].flush()
116 except Exception as e
: # TODO refine
117 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
119 def read(self
, topic
, blocks
=True):
121 Read from one or several topics. it is non blocking returning None if nothing is available
122 :param topic: can be str: single topic; or str list: several topics
123 :param blocks: indicates if it should wait and block until a message is present or returns None
124 :return: topic, key, message; or None if blocks==True
127 if isinstance(topic
, (list, tuple)):
130 topic_list
= (topic
,)
132 for single_topic
in topic_list
:
134 if single_topic
not in self
.files_read
:
135 self
.files_read
[single_topic
] = open(
136 self
.path
+ single_topic
, "a+"
138 self
.buffer[single_topic
] = ""
139 self
.buffer[single_topic
] += self
.files_read
[
142 if not self
.buffer[single_topic
].endswith("\n"):
144 msg_dict
= yaml
.safe_load(self
.buffer[single_topic
])
145 self
.buffer[single_topic
] = ""
146 if len(msg_dict
) != 1:
148 "Length of message dictionary is not equal to 1"
150 for k
, v
in msg_dict
.items():
151 return single_topic
, k
, v
155 except Exception as e
: # TODO refine
156 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
159 self
, topic
, callback
=None, aiocallback
=None, group_id
=None, **kwargs
162 Asyncio read from one or several topics. It blocks
163 :param topic: can be str: single topic; or str list: several topics
164 :param callback: synchronous callback function that will handle the message
165 :param aiocallback: async callback function that will handle the message
166 :param group_id: group_id to use for load balancing. Can be False (set group_id to None), None (use general
167 group_id provided at connect inside config), or a group_id string
168 :param kwargs: optional keyword arguments for callback function
169 :return: If no callback defined, it returns (topic, key, message)
173 msg
= self
.read(topic
, blocks
=False)
176 callback(*msg
, **kwargs
)
178 await aiocallback(*msg
, **kwargs
)
181 await asyncio
.sleep(2)
184 except Exception as e
: # TODO refine
185 raise MsgException(str(e
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
187 async def aiowrite(self
, topic
, key
, msg
):
189 Asyncio write. It blocks
192 :param msg: message, can be str or yaml
193 :return: nothing if ok or raises an exception
195 return self
.write(topic
, key
, msg
)