f731e741fdd801264f86c3c4b97fd630f48c4d88
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
25 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
28 This emulated kafka bus by just using a shared file system. Useful for testing or devops.
29 One file is used per topic. Only one producer and one consumer is allowed per topic. Both consumer and producer
30 access to the same file. e.g. same volume if running with docker.
31 One text line per message is used in yaml format.
35 class MsgLocal(MsgBase
):
37 def __init__(self
, logger_name
='msg'):
38 self
.logger
= logging
.getLogger(logger_name
)
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
):
56 except Exception as e
: # TODO refine
57 raise MsgException(str(e
))
60 for f
in self
.files_read
.values():
63 except Exception: # TODO refine
65 for f
in self
.files_write
.values():
68 except Exception: # TODO refine
71 def write(self
, topic
, key
, msg
):
73 Insert a message into topic
75 :param key: key text to be inserted
76 :param msg: value object to be inserted, can be str, object ...
77 :return: None or raises and exception
80 if topic
not in self
.files_write
:
81 self
.files_write
[topic
] = open(self
.path
+ topic
, "a+")
82 yaml
.safe_dump({key
: msg
}, self
.files_write
[topic
], default_flow_style
=True, width
=20000)
83 self
.files_write
[topic
].flush()
84 except Exception as e
: # TODO refine
85 raise MsgException(str(e
))
87 def read(self
, topic
, blocks
=True):
89 Read from one or several topics. it is non blocking returning None if nothing is available
90 :param topic: can be str: single topic; or str list: several topics
91 :param blocks: indicates if it should wait and block until a message is present or returns None
92 :return: topic, key, message; or None if blocks==True
95 if isinstance(topic
, (list, tuple)):
98 topic_list
= (topic
, )
100 for single_topic
in topic_list
:
101 if single_topic
not in self
.files_read
:
102 self
.files_read
[single_topic
] = open(self
.path
+ single_topic
, "a+")
103 self
.buffer[single_topic
] = ""
104 self
.buffer[single_topic
] += self
.files_read
[single_topic
].readline()
105 if not self
.buffer[single_topic
].endswith("\n"):
107 msg_dict
= yaml
.load(self
.buffer[single_topic
])
108 self
.buffer[single_topic
] = ""
109 assert len(msg_dict
) == 1
110 for k
, v
in msg_dict
.items():
111 return single_topic
, k
, v
115 except Exception as e
: # TODO refine
116 raise MsgException(str(e
))
118 async def aioread(self
, topic
, loop
):
120 Asyncio read from one or several topics. It blocks
121 :param topic: can be str: single topic; or str list: several topics
122 :param loop: asyncio loop
123 :return: topic, key, message
127 msg
= self
.read(topic
, blocks
=False)
130 await asyncio
.sleep(2, loop
=loop
)
133 except Exception as e
: # TODO refine
134 raise MsgException(str(e
))
136 async def aiowrite(self
, topic
, key
, msg
, loop
=None):
138 Asyncio write. It blocks
141 :param msg: message, can be str or yaml
142 :param loop: asyncio loop
143 :return: nothing if ok or raises an exception
145 return self
.write(topic
, key
, msg
)