Added communicators

app.py

@@ -45,23 +45,7 @@ class Scheduler:
         logging.debug('Update redis consumers ip list from first answers: Done')
 
     def request_sync(self, consumer_list_redis):
-        temp_dict = {}
-        for uuid, info in consumer_list_redis.items():
-            ip = info['ip']
-            try:
-                # request synchronization
-                response = requests.post(f"http://{ip}/sync", json={'uuid': os.environ['LOCAL_UUID']}, timeout=5)
-            except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
-                logging.error(f"Error while syncing to {ip}: {str(e)}")
-                continue
 
-            if response.status_code == 200:
-                temp_dict[response.json()['uuid']] = {'ip': ip}
-
-        # update the dictionary of the currently available consumers
-        consumer_list_redis.update(temp_dict)
-        self.r.set('consumer_list', json.dumps(consumer_list_redis).encode('utf-8'))
-        logging.debug('Update redis consumer ip list from answers: Done')
 
     def test_ip_change(self, host_name):
 
@@ -91,16 +75,7 @@ class Scheduler:
                 else:
                     continue
 
-                try:
-                    response = requests.post(
-                        f"http://{ip}/ip",
-                        json={'uuid': os.environ['LOCAL_UUID'], 'ip': current_ip},
-                        timeout=5
-                    )
-                    logging.debug(f"Pushed update to {key.decode('utf-8')} at {ip}. Response: {response.status_code}")
-                except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
-                    logging.warning(f"Could not push update to {key.decode('utf-8')}: {str(e)}")
-                    continue
+
         else:
             logging.debug(f'IP unchanged: {current_ip}')
 
@@ -108,13 +83,6 @@ class Scheduler:
         time.sleep(os.environ.get("RUN_INTERVAL", 30))
 
 
-def get_ip_list(self):
-    # set initial consumer addresses
-    ip_list = os.environ['INITIAL_SERVERS'].split(',')
-    logging.debug('Get consumer list from environ at first: Done')
-    return ip_list
-
-
 def main():
     # set logging preferences
     logging.basicConfig(filename='', level=logging.DEBUG)
@@ -122,27 +90,6 @@ def main():
     redis_storage = RedisSuperStorage(os.environ.get('REDIS_URL', "redis://localhost:6379/0"))
     scheduler = Scheduler()
 
-    # get the dictionary of the currently available consumers
-
-    consumer_list_redis = redis_storage.get_consumer_dictionary()
-    logging.debug('Get consumer list from redis at first: Done')
-
-    ip_list = redis.get_ip_list()
-
-    scheduler.request_list(ip_list)
-    scheduler.request_first_sync(ip_list, consumer_list_redis)
-
-    while True:
-        logging.debug('Infinite Cycle start : Done')
-        # get the dictionary of the currently available consumers
-        consumer_list_redis = redis.get_consumer_dictionary()
-        logging.debug('Get consumer list from redis: Done')
-
-        temp_dict = {}
-
-        scheduler.request_sync(consumer_list_redis)
-        # Test ip change stuff
-        scheduler.test_ip_change(host_name)
 
 
 if __name__ == "__main__":

communicators.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+import os
+import logging
+import requests
+from redis_super_storage import RedisSuperStorage
+
+
+class ProducerCommunicator:
+
+    def __init__(self, redis_store: RedisSuperStorage):
+        self._redis_store = redis_store
+        self._session = requests.Session()
+
+    def push_ip_update(self, newip: str):
+
+        for key, ip in self._redis_store.get_producer_list().items():
+
+            try:
+                response = requests.post(
+                    f"http://{ip}/ip",
+                    json={'uuid': os.environ['LOCAL_UUID'], 'ip': newip},
+                    timeout=5
+                )
+                logging.debug(f"Pushed update to {key} at {ip}. Response: {response.status_code}")
+            except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
+                logging.warning(f"Could not push update to {key}: {str(e)}")
+
+
+class ConsumerCommunicator:
+
+    def __init__(self, redis_store: RedisSuperStorage):
+        self._redis_store = redis_store
+        self._session = requests.Session()
+
+    def targeted_snyc(self, ip: str):
+        try:
+            # request synchronization
+            response = requests.post(f"http://{ip}/sync", json={'uuid': os.environ['LOCAL_UUID']}, timeout=5)
+
+        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
+            logging.error(f"Error while syncing to {ip}: {str(e)}")
+            return
+
+        if response.status_code == 200:
+            self._redis_store.update_consumer(response.json()['uuid'], ip)
+
+    def sync_all(self):
+        for uuid, info in self._redis_store.get_consumer_list().items():
+            ip = info['ip']
+            try:
+                # request synchronization
+                response = requests.post(f"http://{ip}/sync", json={'uuid': os.environ['LOCAL_UUID']}, timeout=5)
+
+            except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
+                logging.error(f"Error while syncing to {ip}: {str(e)}")
+                continue
+
+            if response.status_code == 200:
+                self._redis_store.update_consumer(response.json()['uuid'], ip)

redis_super_storage.py

@@ -24,16 +24,16 @@ class RedisSuperStorage:
 
         return list_of_customers
 
-    def get_producer_list(self) -> list:
+    def get_producer_list(self) -> dict:
         keys = self.r.keys('producer_*')
 
-        list_of_producer_ip = []
+        list_of_producer_ip = {}
 
         for key in keys:
             ip = (self.r.get(key) or b"").decode('utf-8')
 
             if ip:
-                list_of_producer_ip.append(ip)
+                list_of_producer_ip[key.decode('utf-8')] = ip
 
         return list_of_producer_ip