109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
#!/usr/bin/env python
|
|
|
|
"""
|
|
Communicator module
|
|
"""
|
|
|
|
import logging
|
|
import requests
|
|
import requests.exceptions
|
|
from consumerinformation import ConsumerInformation
|
|
|
|
__author__ = "@tormakris"
|
|
__copyright__ = "Copyright 2020, GoldenPogácsa Team"
|
|
__module_name__ = "messagesender"
|
|
__version__text__ = "1"
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class Communicator:
|
|
"""
|
|
Class handling low level communication with consumers.
|
|
"""
|
|
|
|
def __init__(self, currentconsumer: str, uuid: str, consumerinformation: ConsumerInformation):
|
|
"""**Constructor:**
|
|
Initializes the object.
|
|
|
|
:param currentconsumer: the current consumer's IP address as a string
|
|
:param uuid: string typed UUID.
|
|
"""
|
|
self.currenctconsumer = currentconsumer
|
|
self.uuid = uuid
|
|
self.consumerinformation = consumerinformation
|
|
|
|
def sendmessage(self, message: str) -> None:
|
|
"""Send message to the current consumer. Logs the process.
|
|
|
|
:param message: the message of type string that will be sent.
|
|
:return: None
|
|
"""
|
|
currentconsumer = self.currenctconsumer
|
|
LOGGER.info(f"Sending message to {currentconsumer}")
|
|
|
|
for consumer in self.consumerinformation.getconsumerlist():
|
|
try:
|
|
postresponse = requests.post(f'http://{consumer}/log', json={'uuid': self.uuid, 'message': message},
|
|
timeout=5)
|
|
LOGGER.debug(f"Message status code is:{postresponse.status_code}")
|
|
if postresponse.status_code < 300:
|
|
return None
|
|
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
|
|
LOGGER.warning(f"Could not send message to {consumer}")
|
|
|
|
def discoveravailableconsumers(self) -> list:
|
|
"""Get the list of available consumer from the current primary consumer. Logs the received list.
|
|
|
|
:return: list of consumers' IP addresses
|
|
"""
|
|
try:
|
|
currentconsumer = self.currenctconsumer
|
|
response = requests.get(f'http://{currentconsumer}/consumers', timeout=5)
|
|
json = response.json()
|
|
LOGGER.info(f"List of currently available consumers: {json}")
|
|
return json
|
|
except Exception as e:
|
|
LOGGER.warning("Could not query available consumer list!")
|
|
# LOGGER.exception(e)
|
|
return []
|
|
|
|
def isconsumeravailable(self) -> bool:
|
|
"""Readiness probe current consumer. Logs the result.
|
|
|
|
:return: True if available, False otherwise
|
|
"""
|
|
currentconsumer = self.currenctconsumer
|
|
try:
|
|
response = requests.get(f'http://{currentconsumer}/consumers', timeout=5)
|
|
isavailable = response.status_code == 200
|
|
except Exception as e:
|
|
# LOGGER.exception(e)
|
|
isavailable = False
|
|
LOGGER.info(f"Current consumer availability: {isavailable}")
|
|
return isavailable
|
|
|
|
def checkconsumer(self, consumer: str) -> bool:
|
|
"""Readiness probe of a particular consumer. Logs the result.
|
|
|
|
:param consumer: the consumer's IP address
|
|
:return: True if available, False otherwise
|
|
"""
|
|
try:
|
|
response = requests.get(f'http://{consumer}/consumers', timeout=5)
|
|
isavailable = response.status_code == 200
|
|
except Exception as e:
|
|
# LOGGER.exception(e)
|
|
isavailable = False
|
|
LOGGER.info(f"Consumer {consumer} availability: {isavailable}")
|
|
return isavailable
|
|
|
|
def set_currentconsumer(self, currenctconsumer) -> None:
|
|
"""Set current consumer
|
|
|
|
:param currenctconsumer: the consumer's IP address
|
|
:return: None
|
|
"""
|
|
self.currenctconsumer = currenctconsumer
|