1 # -*- coding: utf-8 -*-
3 # Copyright 2018 Whitestack, LLC
4 # *************************************************************
6 # This file is part of OSM Monitoring module
7 # All Rights Reserved to Whitestack, LLC
9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
10 # not use this file except in compliance with the License. You may obtain
11 # a copy of the License at
13 # http://www.apache.org/licenses/LICENSE-2.0
15 # Unless required by applicable law or agreed to in writing, software
16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
18 # License for the specific language governing permissions and limitations
21 # For those usages not covered by the Apache License, Version 2.0 please
22 # contact: bdiaz@whitestack.com or glavado@whitestack.com
25 from typing
import List
, Callable
27 from osm_common
import msgkafka
, msglocal
29 from osm_mon
.core
.config
import Config
32 class MessageBusClient
:
33 def __init__(self
, config
: Config
, loop
=None):
34 if config
.get('message', 'driver') == "local":
35 self
.msg_bus
= msglocal
.MsgLocal()
36 elif config
.get('message', 'driver') == "kafka":
37 self
.msg_bus
= msgkafka
.MsgKafka()
39 raise Exception("Unknown message bug driver {}".format(config
.get('section', 'driver')))
40 self
.msg_bus
.connect(config
.get('message'))
42 loop
= asyncio
.get_event_loop()
45 async def aioread(self
, topics
: List
[str], callback
: Callable
= None, **kwargs
):
47 Retrieves messages continuously from bus and executes callback for each message consumed.
48 :param topics: List of message bus topics to consume from.
49 :param callback: Async callback function to be called for each message received.
50 :param kwargs: Keyword arguments to be passed to callback function.
53 await self
.msg_bus
.aioread(topics
, self
.loop
, aiocallback
=callback
, **kwargs
)
55 async def aiowrite(self
, topic
: str, key
: str, msg
: dict):
57 Writes message to bus.
58 :param topic: Topic to write to.
59 :param key: Key to write to.
60 :param msg: Dictionary containing message to be written.
63 await self
.msg_bus
.aiowrite(topic
, key
, msg
, self
.loop
)
65 async def aioread_once(self
, topic
: str):
67 Retrieves last message from bus.
68 :param topic: topic to retrieve message from.
69 :return: tuple(topic, key, message)
71 result
= await self
.msg_bus
.aioread(topic
, self
.loop
)