Issue 133 and 134 merge branch

__init__.py

@@ -0,0 +1,4 @@
+import sys
+import os
+
+sys.path.append(os.path.abspath(os.path.dirname(__file__)))

onair/data/telemetry_configs/redis_example_CONFIG.json

@@ -24,11 +24,25 @@
       }
     }
   },
-  "redis_subscriptions": [
-    "state_0",
-    "state_1",
-    "state_2"
-  ],
+
+  "redis" : [
+      {
+        "address": "localhost",
+        "port": 6379,
+        "subscriptions": [
+          "state_0"
+        ]
+      },
+      {
+        "address": "localhost",
+        "port": 6380,
+        "subscriptions": [
+          "state_1",
+          "state_2"
+        ]
+      }
+    ],
+
   "order": [
     "time",
     "state_0.x",

onair/data_handling/redis_adapter.py

@@ -29,60 +29,96 @@ class DataSource(OnAirDataSource):
 
     def __init__(self, data_file, meta_file, ss_breakdown = False):
         super().__init__(data_file, meta_file, ss_breakdown)
-        self.address = 'localhost'
-        self.port = 6379
-        self.db = 0
-        self.server = None
         self.new_data_lock = threading.Lock()
         self.new_data = False
+        self.servers = []
         self.currentData = []
         self.currentData.append({'headers':self.order,
                                  'data':list('-' * len(self.order))})
         self.currentData.append({'headers':self.order,
                                  'data':list('-' * len(self.order))})
         self.double_buffer_read_index = 0
         self.connect()
-        self.subscribe(self.subscriptions)
 
     def connect(self):
         """Establish connection to REDIS server."""
         print_msg('Redis adapter connecting to server...')
-        self.server = redis.Redis(self.address, self.port, self.db)
+        for idx, server_config in enumerate(self.server_configs):
+            server_config_keys = server_config.keys()
+            if 'address' in server_config_keys:
+                address = server_config['address']
+            else:
+                address = 'localhost'
 
-        if self.server.ping():
-            print_msg('... connected!')
+            if 'port' in server_config_keys:
+                port = server_config['port']
+            else:
+                port = 6379
+
+            if 'db' in server_config_keys:
+                db = server_config['db']
+            else:
+                db = 0
 
-    def subscribe(self, subscriptions):
-        """Subscribe to REDIS message channel(s) and launch listener thread."""
-        if len(subscriptions) != 0 and self.server.ping():
-            self.pubsub = self.server.pubsub()
+            if 'password' in server_config_keys:
+                password = server_config['password']
+            else:
+                password = ''
+
+            #if there are subscriptions in this Redis server configuration's subscription key
+            if len(server_config['subscriptions']) != 0:
+                #Create the servers and append them to self.servers list
+                self.servers.append(redis.Redis(address, port, db, password))
 
-            for s in subscriptions:
-                self.pubsub.subscribe(s)
-                print_msg(f"Subscribing to channel: {s}")
+                try:
+                    #Ping server to make sure we can connect
+                    self.servers[-1].ping()
+                    print_msg(f'... connected to server # {idx}!')
 
-            listen_thread = threading.Thread(target=self.message_listener)
-            listen_thread.start()
-        else:
-            print_msg(f"No subscriptions given!")
+                    #Set up Redis pubsub function for the current server
+                    pubsub = self.servers[-1].pubsub()
+
+                    for s in server_config['subscriptions']:
+                        pubsub.subscribe(s)
+                        print_msg(f"Subscribing to channel: {s} on server # {idx}")
+                    listen_thread = threading.Thread(target=self.message_listener, args=(pubsub,))
+                    listen_thread.start()
+
+                #This except will be hit if self.servers[-1].ping() threw an exception (could not properly ping server)
+                except:
+                    print_msg(f'Did not connect to server # {idx}. Not setting up subscriptions.', 'RED')
+
+            else:
+                print_msg("No subscriptions given! Redis server not created")       
 
     def parse_meta_data_file(self, meta_data_file, ss_breakdown):
+        self.server_configs = []
         configs = extract_meta_data_handle_ss_breakdown(
             meta_data_file, ss_breakdown)
         meta = parseJson(meta_data_file)
         keys = meta.keys()
 
+        # Setup redis server configuration
+        #Checking if 'redis' exists
+        if 'redis' in keys:
+            count_server_config = 0
+            #Checking if dictionaries within 'redis' key each have a 'subscription' key. Error will be thrown if not.
+            for server_config in meta['redis']:
+                redis_config_keys = server_config.keys()
+                if ('subscriptions' in redis_config_keys) == False:
+                    raise ConfigKeyError(f'Config file: \'{meta_data_file}\' ' \
+                        f'missing required key \'subscriptions\' from {count_server_config} in key \'redis\'')  
+                count_server_config +=1
+
+            #Saving all of Redis dictionaries from JSON file to self.server_configs
+            self.server_configs = meta['redis']
+
         if 'order' in keys:
             self.order = meta['order']
         else:
             raise ConfigKeyError(f'Config file: \'{meta_data_file}\' ' \
-                                  'missing required key \'order\'')
-
-        if 'redis_subscriptions' in meta.keys():
-            self.subscriptions = meta['redis_subscriptions']
-        else:
-            self.subscriptions = []
-
+                                'missing required key \'order\'')
+
         return configs
 
     def process_data_file(self, data_file):
@@ -115,9 +151,9 @@ def has_more(self):
         """Live connection should always return True"""
         return True
 
-    def message_listener(self):
+    def message_listener(self, pubsub):
         """Loop for listening for messages on channels"""
-        for message in self.pubsub.listen():
+        for message in pubsub.listen():
             if message['type'] == 'message':
                 channel_name = f"{message['channel'].decode()}"
                 # Attempt to load message as json
@@ -173,4 +209,4 @@ def message_listener(self):
         print_msg("Redis subscription listener exited.", ['WARNING'])
 
     def has_data(self):
-        return self.new_data
+        return self.new_data

redis-experiment-publisher-multi-server.py

@@ -0,0 +1,70 @@
+import redis
+import time
+import random
+
+# Initialize the Redis connection for server #1
+redis_host = "localhost"
+redis_port = 6379
+# When your Redis server requires a password, fill it in here
+redis_password = ""
+# Connect to Redis
+r1 = redis.Redis(host=redis_host,
+                port=redis_port,
+                password=redis_password,
+                decode_responses=True)
+
+# Initialize the Redis connection for server #1
+redis_host = "localhost"
+redis_port = 6380
+# When your Redis server requires a password, fill it in here
+redis_password = ""
+# Connect to Redis
+r2 = redis.Redis(host=redis_host,
+                port=redis_port,
+                password=redis_password,
+                decode_responses=True)
+
+# List of channel names
+server1_channels = ['state_0']
+server2_channels = ['state_1', 'state_2']
+# Publish messages on each channel in random order
+def publish_messages():
+    loop_count = 0
+    inner_loop_count = 0
+    max_loops = 9
+    while loop_count < max_loops:
+        random.shuffle(server1_channels)
+        for channel in server1_channels:
+            r1.publish(channel, f'{{"time":{inner_loop_count}, ' \
+                               f'"x":{inner_loop_count+0.1}, ' \
+                               f'"y":{inner_loop_count+0.2}}}')           
+
+            print(f"Published data to {channel}, " \
+                  f"[{inner_loop_count}, " \
+                  f"{inner_loop_count+0.1}, " \
+                  f"{inner_loop_count+0.2}]")            
+
+            inner_loop_count += 1
+            time.sleep(2)
+
+        random.shuffle(server2_channels)
+        for channel in server2_channels:           
+            r2.publish(channel, f'{{"time":{inner_loop_count}, ' \
+                               f'"x":{inner_loop_count+0.1}, ' \
+                               f'"y":{inner_loop_count+0.2}}}')
+
+            print(f"Published data to {channel}, " \
+                  f"[{inner_loop_count}, " \
+                  f"{inner_loop_count+0.1}, " \
+                  f"{inner_loop_count+0.2}]")            
+
+            inner_loop_count += 1
+            time.sleep(2)
+
+        loop_count += 1
+        print(f"Completed {loop_count} loops")
+
+
+
+if __name__ == "__main__":
+    publish_messages()