Added functions for active listening

broker_connect.py

@@ -1,3 +1,4 @@
+import threading
 import json
 import time
 
@@ -60,6 +61,18 @@ def on_message(self, _client, _userdata, msg):
         """Update message queue with latest broker response."""
         self.last_message = json.loads(msg.payload)
 
+    def show_connect(self, client, *_args):
+        # Subscribe to all channels
+        client.subscribe(f"bot/{self.token['token']['unencoded']['bot']}/#")
+
+    def show_message(self, _client, _userdata, msg):
+        self.last_message = msg.payload
+        print('-' * 100)
+        # Print channel
+        print(f'{msg.topic} ({datetime.now().strftime("%Y-%m-%d %H:%M:%S")})\n')
+        # Print message
+        print(json.dumps(json.loads(msg.payload), indent=4))
+
     def listen(self, duration, channel):
         """Listen to messages via message broker."""
 
@@ -77,3 +90,21 @@ def listen(self, duration, channel):
 
         self.client.loop_stop()
         self.client.disconnect()
+
+    def start_listening(self):
+        print("Now listening to message broker...")
+
+        if self.client is None:
+            self.connect()
+
+        # Wrap on_connect to pass channel argument
+        self.client.on_connect = self.show_connect
+        self.client.on_message = self.show_message
+
+        self.client.loop_start()
+
+    def stop_listening(self):
+        print("Stopped listening to message broker...")
+
+        self.client.loop_stop()
+        self.client.disconnect()

broker_functions.py

@@ -604,7 +604,7 @@ def get_job(self, job_str=None):
         # Return job as json object: job[""]
         return jobs
 
-    def set_job(self, job_str, field=None, new_val=None):
+    def set_job(self, job_str, field=None, new_val=None): # TODO: implement changes via message broker
         # Add new or edit single by name
         status_data = self.read_status()
         jobs = status_data["jobs"]
@@ -634,7 +634,7 @@ def set_job(self, job_str, field=None, new_val=None):
         # Return job as json object: job[""]
         return None
 
-    def complete_job(self, job_str):
+    def complete_job(self, job_str): # TODO: implement changes via message broker
         status_data = self.read_status()
         jobs = status_data["jobs"]