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
24 from typing
import List
, Callable
26 from osm_common
import msgkafka
, msglocal
28 from osm_policy_module
.core
.config
import Config
31 class MessageBusClient
:
32 def __init__(self
, config
: Config
):
33 if config
.get("message", "driver") == "local":
34 self
.msg_bus
= msglocal
.MsgLocal()
35 elif config
.get("message", "driver") == "kafka":
36 self
.msg_bus
= msgkafka
.MsgKafka()
39 "Unknown message bug driver {}".format(config
.get("section", "driver"))
41 self
.msg_bus
.connect(config
.get("message"))
43 async def aioread(self
, topics
: List
[str], callback
: Callable
= None, **kwargs
):
45 Retrieves messages continuously from bus and executes callback for each message consumed.
46 :param topics: List of message bus topics to consume from.
47 :param callback: Async callback function to be called for each message received.
48 :param kwargs: Keyword arguments to be passed to callback function.
51 await self
.msg_bus
.aioread(topics
, aiocallback
=callback
, **kwargs
)
53 async def aiowrite(self
, topic
: str, key
: str, msg
: dict):
55 Writes message to bus.
56 :param topic: Topic to write to.
57 :param key: Key to write to.
58 :param msg: Dictionary containing message to be written.
61 await self
.msg_bus
.aiowrite(topic
, key
, msg
)
63 async def aioread_once(self
, topic
: str):
65 Retrieves last message from bus.
66 :param topic: topic to retrieve message from.
67 :return: tuple(topic, key, message)
69 result
= await self
.msg_bus
.aioread(topic
)