app.py

@@ -10,7 +10,6 @@ import uuid
 import logging
 import sentry_sdk
 import time
-from consumerinformation import ConsumerInformation
 from communicator import Communicator
 from consumerlocator import ConsumerLocator
 from messagesender import MessageSender
@@ -43,14 +42,11 @@ if __name__ == "__main__":
     """
     LOGGER.info("Producer started")
     generateduuid = str(uuid.uuid4())
-    redisconnector = RedisConnector()
+    communicator = Communicator(currentconsumer=KNOWNCONSUMER, uuid=generateduuid)
-    consumerinfomation = ConsumerInformation(redisconnector=redisconnector)
-    communicator = Communicator(currentconsumer=KNOWNCONSUMER, uuid=generateduuid,
-                                consumerinformation=consumerinfomation)
     LOGGER.debug(f"My uuid is {generateduuid}")
     messagesender = MessageSender(communicator=communicator, uuid=generateduuid)
-    consumerlocator = ConsumerLocator(communicator=communicator,
+    consumerlocator = ConsumerLocator(uuid=generateduuid, communicator=communicator,
-                                      redisconnector=redisconnector)
+                                      redisconnector=RedisConnector())
 
     while True:
         consumerlocator.learnconsumerlist()

communicator.py

@@ -7,7 +7,6 @@ Communicator module
 import logging
 import requests
 import requests.exceptions
-from consumerinformation import ConsumerInformation
 
 __author__ = "@tormakris"
 __copyright__ = "Copyright 2020, GoldenPogácsa Team"
@@ -23,16 +22,15 @@ class Communicator:
     Class handling low level communication with consumers.
     """
 
-    def __init__(self, currentconsumer: str, uuid: str, consumerinformation: ConsumerInformation):
+    def __init__(self, currentconsumer: str, uuid: str):
         """**Constructor:**
         Initializes the object.
 
-        :param currentconsumer: the current consumer's IP address as a string
+        :param consumerlocator: the current consumer's IP address as a string
         :param uuid: string typed UUID.
         """
-        self.currenctconsumer = currentconsumer
+        self.currenctconsumer=currentconsumer
         self.uuid = uuid
-        self.consumerinformation = consumerinformation
 
     def sendmessage(self, message: str) -> None:
         """Send message to the current consumer. Logs the process.
@@ -40,18 +38,13 @@ class Communicator:
         :param message: the message of type string that will be sent.
         :return: None
         """
-        currentconsumer = self.currenctconsumer
+        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},
+            postresponse=requests.post(f'http://{currentconsumer}/log', json={'uuid': self.uuid, 'message': message}, timeout=5)
-                                             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}")
+            LOGGER.exception(e)  # Fun fact: ez azt jelenti, hogy elveszett az üzenet... ide valami retry kellene inkább más consumerek felé...
 
     def discoveravailableconsumers(self) -> list:
         """Get the list of available consumer from the current primary consumer. Logs the received list.
@@ -65,8 +58,8 @@ class Communicator:
             LOGGER.info(f"List of currently available consumers: {json}")
             return json
         except Exception as e:
-            LOGGER.warning("Could not query available consumer list!")
+            LOGGER.error("Could not query available consumer list!")
-            # LOGGER.exception(e)
+            #LOGGER.exception(e)
             return []
 
     def isconsumeravailable(self) -> bool:
@@ -79,7 +72,7 @@ class Communicator:
             response = requests.get(f'http://{currentconsumer}/consumers', timeout=5)
             isavailable = response.status_code == 200
         except Exception as e:
-            # LOGGER.exception(e)
+            #LOGGER.exception(e)
             isavailable = False
         LOGGER.info(f"Current consumer availability: {isavailable}")
         return isavailable
@@ -94,15 +87,15 @@ class Communicator:
             response = requests.get(f'http://{consumer}/consumers', timeout=5)
             isavailable = response.status_code == 200
         except Exception as e:
-            # LOGGER.exception(e)
+            #LOGGER.exception(e)
             isavailable = False
         LOGGER.info(f"Consumer {consumer} availability: {isavailable}")
         return isavailable
 
-    def set_currentconsumer(self, currenctconsumer) -> None:
+    def set_currentconsumer(self,currenctconsumer) -> None:
         """Set current consumer
 
         :param currenctconsumer: the consumer's IP address
         :return: None
         """
-        self.currenctconsumer = currenctconsumer
+        self.currenctconsumer=currenctconsumer

consumerinformation.py

@@ -1,37 +0,0 @@
-#!/usr/bin/env python
-
-"""
-Consumer locator module, that manages the list of consumers.
-"""
-
-from redisconnector import RedisConnector
-
-__author__ = "@ricsik52"
-__copyright__ = "Copyright 2020, GoldenPogácsa Team"
-__module_name__ = "consumerlocator"
-__version__text__ = "1"
-
-
-class ConsumerInformation:
-    """
-    Component responsible for providing high level information about consumers
-    """
-
-    def __init__(self, redisconnector: RedisConnector):
-        """**Constructor:**
-        Initializes the object.
-
-        :param redisconnector: the :class:'redisconnector.RedisConnector' instance that will be used for Redis connection.
-        """
-        self.red = redisconnector
-
-    def getconsumerlist(self) -> list:
-        """
-        Gets the list of currently available consumers.
-        :return:
-        """
-        toreturn = []
-        for consumer in self.red.consumerlist:
-            if consumer['State']:
-                toreturn.append(consumer['Host'])
-        return toreturn

consumerlocator.py

@@ -20,16 +20,18 @@ LOGGER = logging.getLogger(__name__)
 
 
 class ConsumerLocator:
+
     """
     Component responsible for managing the list of consumers. Requires an instance of :class:`communicator.Communicator`
     """
 
-    def __init__(self, communicator: Communicator, redisconnector: RedisConnector):
+    def __init__(self, uuid: str, communicator: Communicator, redisconnector: RedisConnector):
         """**Constructor:**
         Initializes the object.
 
         Gets the known consumer's IP address from the PRODUCER_KNOWNCONSUMER envar.
 
+        :param uuid: Not used
         :param communicator: the :class:'communicator.Communicator' instance that will be used for the low level communication.
         """
         self.red = redisconnector
@@ -49,6 +51,7 @@ class ConsumerLocator:
         if not recievedconsumerlist:
             return
 
+
         consumer_list = self.red.consumerlist
 
         for recconsumer in recievedconsumerlist:
@@ -59,8 +62,7 @@ class ConsumerLocator:
 
             if not contains:
                 LOGGER.info(f"Learned about new consumer at {recconsumer}")
-                consumer_list.append(
+                consumer_list.append({"Host": recconsumer, "State": True, "LastOk": datetime.datetime.now().timestamp()})
-                    {"Host": recconsumer, "State": True, "LastOk": datetime.datetime.now().timestamp()})
 
         self.red.consumerlist = consumer_list
         self.updateconsumerlist()
@@ -79,8 +81,7 @@ class ConsumerLocator:
         for consumer in consumer_list:
             if not self.communicator.checkconsumer(consumer["Host"]):
                 consumer["State"] = False
-                if datetime.datetime.now() - datetime.datetime.fromtimestamp(consumer["LastOk"]) > datetime.timedelta(
+                if datetime.datetime.now() - datetime.datetime.fromtimestamp(consumer["LastOk"]) > datetime.timedelta(hours=1):
-                        hours=1):
                     removelist.append(consumer)
             else:
                 consumer["LastOk"] = datetime.datetime.now().timestamp()

integtest.py

@@ -4,8 +4,9 @@
 Integration test for the producer module.
 """
 
+import os
 import re
-from consumerinformation import ConsumerInformation
+import datetime
 from communicator import Communicator
 from consumerlocator import ConsumerLocator
 from messagesender import MessageSender
@@ -22,14 +23,11 @@ redis_proc = factories.redis_proc(host='cache', port=6379)
 redis_db = factories.redisdb('redis_nooproc')
 
 answer = ""
-
-
 def answermethod(Request):
     global answer
     answer = Request
     return ""
 
-
 def test_integration(httpserver):
     """
     Tests the whole system.
@@ -47,14 +45,12 @@ def test_integration(httpserver):
         method='POST').respond_with_handler(answermethod)
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnector = RedisConnector()
-    consumerinfomation = ConsumerInformation(redisconnector=redisconnector)
     comm = Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfomation)
+        uuid=generateduuid)
     messagesender = MessageSender(communicator=comm, uuid=generateduuid)
-    consumerlocator = ConsumerLocator(communicator=comm,
+    consumerlocator = ConsumerLocator(uuid=generateduuid, communicator=comm,
-                                      redisconnector=redisconnector)
+                                      redisconnector=RedisConnector())
     #First test method
     consumerlocator.learnconsumerlist()
     conslist = consumerlocator.red.consumerlist

test.py

@@ -9,7 +9,6 @@ import consumerlocator
 import communicator
 import messagesender
 import redisconnector
-import consumerinformation
 from pytest_redis import factories
 
 __author__ = "@tormakris"
@@ -29,11 +28,9 @@ def test_generate_string(mocker):
     :param mocker: patches the :class:`communicator.Communicator`.
     """
     mocker.patch('communicator.Communicator')
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
     comm = communicator.Communicator(
         currentconsumer="localhost",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     mess = messagesender.MessageSender(communicator=comm, uuid=generateduuid)
     msg = mess.randomstring(stringlength=32)
     assert isinstance(msg, str)
@@ -55,13 +52,9 @@ def test_sendmessage(httpserver):
             "test": "ok"})
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     comm = communicator.Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     mess = "SENDING"
     ret = comm.sendmessage(message=mess)
     assert ret is None
@@ -75,13 +68,9 @@ def test_send_message(mocker):
     :return: None
     """
     mocker.patch('communicator.Communicator')
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1", "State": True, "LastOk": "1589479202"}]
     comm = communicator.Communicator(
         currentconsumer="127.0.0.1",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     mess = messagesender.MessageSender(communicator=comm, uuid=generateduuid)
     messa = "SENDING"
     msg = mess.sendmessage(message=messa)
@@ -102,13 +91,9 @@ def test_discoveravailableconsumers(httpserver):
         ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     comm = communicator.Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     ret = comm.discoveravailableconsumers()
     assert isinstance(ret, list)
     assert ret == ["10.69.42.1", "10.10.10.10", "10.20.30.40"]
@@ -128,13 +113,9 @@ def test_isconsumeravailable(httpserver):
         ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     comm = communicator.Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     ret = comm.isconsumeravailable()
     assert isinstance(ret, bool)
     assert ret
@@ -145,7 +126,7 @@ def test_isconsumeravailable(httpserver):
 
     comm2 = communicator.Communicator(
         currentconsumer="127.0.0.1:69",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
 
     ret3 = comm2.isconsumeravailable()
     assert isinstance(ret3, bool)
@@ -166,13 +147,9 @@ def test_checkconsumer(httpserver):
         ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     comm = communicator.Communicator(
         currentconsumer="127.0.0.1",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     ret = comm.checkconsumer(f"127.0.0.1:{port}")
     assert isinstance(ret, bool)
     assert ret
@@ -183,7 +160,8 @@ def test_checkconsumer(httpserver):
 
     comm2 = communicator.Communicator(
         currentconsumer="127.0.0.1",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
+
     ret3 = comm2.checkconsumer(f"127.0.0.1:{port}")
     assert isinstance(ret3, bool)
     assert ret3 == False
@@ -195,11 +173,9 @@ def test_setcurrentconsumer():
 
     :return: None
     """
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
     comm = communicator.Communicator(
         currentconsumer="127.0.0.1",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     comm.set_currentconsumer("10.69.42.1")
     assert comm.currenctconsumer == "10.69.42.1"
 
@@ -218,16 +194,12 @@ def test_learnconsumerlist(httpserver):
         ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     comm = communicator.Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
     locator = consumerlocator.ConsumerLocator(
-        communicator=comm,
+        uuid=generateduuid, communicator=comm,
         redisconnector=redisconnector.RedisConnector())
     ret = locator.learnconsumerlist()
     assert ret is None
@@ -241,13 +213,11 @@ def test_getcurrentconsumer(mocker):
     :return: None
     """
     mocker.patch('communicator.Communicator')
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
     comm = communicator.Communicator(
         currentconsumer="127.0.0.1",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     locator = consumerlocator.ConsumerLocator(
-        communicator=comm,
+        uuid=generateduuid, communicator=comm,
         redisconnector=redisconnector.RedisConnector())
     assert locator.getcurrentconsumer() == consumerlocator.KNOWNCONSUMER
 
@@ -266,17 +236,13 @@ def test_checkcurrentconsumer(httpserver):
         ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
     comm = communicator.Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
     locator = consumerlocator.ConsumerLocator(
-        communicator=comm,
+        uuid=generateduuid, communicator=comm,
         redisconnector=redisconnector.RedisConnector())
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     ret = locator.checkcurrentconsumer()
     assert ret == True
 
@@ -295,18 +261,14 @@ def test_updateconsumer(httpserver):
         ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
     comm = communicator.Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
     redisconn = redisconnector.RedisConnector()
     locator = consumerlocator.ConsumerLocator(
-        communicator=comm,
+        uuid=generateduuid, communicator=comm,
         redisconnector=redisconn)
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     assert redisconn.currentconsumer is not None
     ret = locator.updateconsumer()
     assert ret == f"127.0.0.1:{port}"
@@ -326,16 +288,12 @@ def test_updateconsumerlist(httpserver):
         ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
     url = httpserver.url_for("/")
     port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
-    redisconnect = redisconnector.RedisConnector()
-    consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
     comm = communicator.Communicator(
         currentconsumer=f"127.0.0.1:{port}",
-        uuid=generateduuid, consumerinformation=consumerinfo)
+        uuid=generateduuid)
     consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
     locator = consumerlocator.ConsumerLocator(
-        communicator=comm,
+        uuid=generateduuid, communicator=comm,
         redisconnector=redisconnector.RedisConnector())
-    redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
-    redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
     ret = locator.updateconsumerlist()
     assert ret is None