X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_policy_module%2Fcommon%2Flcm_client.py;h=2efa241a0ac009367f6211828c7914b4d71d362f;hb=refs%2Fchanges%2F78%2F7378%2F1;hp=34e212f7fdf722a15024076d0333c2d73b1cb3c1;hpb=4c5541f020bbc56e9ae1a326b0d33e673dcd5f0b;p=osm%2FPOL.git diff --git a/osm_policy_module/common/lcm_client.py b/osm_policy_module/common/lcm_client.py index 34e212f..2efa241 100644 --- a/osm_policy_module/common/lcm_client.py +++ b/osm_policy_module/common/lcm_client.py @@ -21,40 +21,34 @@ # For those usages not covered by the Apache License, Version 2.0 please # contact: bdiaz@whitestack.com or glavado@whitestack.com ## +import asyncio import datetime import json import logging import time import uuid -from kafka import KafkaProducer -from osm_common import dbmongo - +from osm_policy_module.common.common_db_client import CommonDbClient +from osm_policy_module.common.message_bus_client import MessageBusClient from osm_policy_module.core.config import Config log = logging.getLogger(__name__) class LcmClient: - def __init__(self): - cfg = Config.instance() - self.kafka_server = '{}:{}'.format(cfg.OSMPOL_MESSAGE_HOST, - cfg.OSMPOL_MESSAGE_PORT) - self.producer = KafkaProducer(bootstrap_servers=self.kafka_server, - key_serializer=str.encode, - value_serializer=str.encode) - self.common_db = dbmongo.DbMongo() - self.common_db.db_connect({'host': cfg.OSMPOL_DATABASE_HOST, - 'port': int(cfg.OSMPOL_DATABASE_PORT), - 'name': 'osm'}) + def __init__(self, config: Config, loop=None): + self.db_client = CommonDbClient(config) + self.msg_bus = MessageBusClient(config) + if not loop: + loop = asyncio.get_event_loop() + self.loop = loop - def scale(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str): + async def scale(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str): log.debug("scale %s %s %s %s", nsr_id, scaling_group_name, vnf_member_index, action) nslcmop = self._generate_nslcmop(nsr_id, scaling_group_name, vnf_member_index, action) - self.common_db.create("nslcmops", nslcmop) + self.db_client.create_nslcmop(nslcmop) log.info("Sending scale action message: %s", json.dumps(nslcmop)) - self.producer.send(topic='ns', key='scale', value=json.dumps(nslcmop)) - self.producer.flush() + await self.msg_bus.aiowrite("ns", "scale", nslcmop) def _generate_nslcmop(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str): log.debug("_generate_nslcmop %s %s %s %s", nsr_id, scaling_group_name, vnf_member_index, action)