fix publisher and subscriber, new paho-mqtt version

code/eval/queries/cypher/driver_latest/cypher_queries.py

@@ -66,16 +66,23 @@ def execute_and_log_query(query_key, query):
     start_time = time.time()
     with driver.session() as session:
         result = session.run(query)
+        row_count= len(result.data())
         summary = result.consume()
         end_time = time.time()
 
     # Check if the attributes are None and set them to 0 if they are
     result_available_after = summary.result_available_after if summary.result_available_after is not None else 0
     result_consumed_after = summary.result_consumed_after if summary.result_consumed_after is not None else 0
+    row_count = row_count if row_count is not None else 0 
     execution_time = result_available_after + result_consumed_after
 
     logging.info(f"Results for Query {query_key}:")
     logging.info(f"Python executed in: {end_time - start_time}s")
+    
+    if row_count == 0:
+        logging.info(f"Number of rows not available")
+    else:
+        logging.info(f"Number of rows {row_count}")
     if execution_time == 0:
         logging.warning(f"No internal DBMS metric available")
     else:
@@ -86,8 +93,8 @@ def execute_and_log_query(query_key, query):
     # Write a header if the file is newly created or empty, else append the data
     csv_log_file.seek(0, 2)  # Move the cursor to the end of the file
     if csv_log_file.tell() == 0:  # If file is empty, write a header
-        writer.writerow(['Query Key', 'Start Time', 'End Time', 'Execution Time (ms)', 'Total Time (s)'])
-    writer.writerow([query_key, start_time, end_time, execution_time, end_time - start_time])
+        writer.writerow(['Query Key', 'Start Time', 'End Time','Fetched nodes', 'Execution Time (ms)', 'Total Time (s)'])
+    writer.writerow([query_key, start_time, end_time, row_count, execution_time, end_time - start_time])
 
 # Function to schedule and execute all queries
 def schedule_and_execute_queries():

code/infrastructure/streaming/clients/pub/pub_cdm.py

@@ -23,7 +23,7 @@ files = [
     'ta1-cadets-e3-official-2_1.json'
     ]
 line_count = [4999999,4999999,3911712,4999999,4999999,4999999,4999999,4999999,2059063,4999999,34335661] # line_count corresponding to each file
-client = mqtt.Client("Client1")
+client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1,"Client1")
 topic = "neo4j"
 
 def read_moving_window_and_send_data(file_path, lines_per_window, line_count):

code/infrastructure/streaming/clients/sub/memgraph/sub_mem.py

@@ -6,7 +6,7 @@ import os
 
 broker_hostname=str(os.getenv('mos_host',default="localhost"))
 broker_port = int(os.getenv('mos_port',default=1883)) 
-client = mqtt.Client("Client4")
+client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1,"Client4")
 db_uri =str(os.getenv('mem_host',default="bolt://localhost:7687"))
 neo4j_auth=("","")
 abort_time_limit = int(os.getenv('abort_time_limit', default=99999))

code/infrastructure/streaming/clients/sub/neo4j/sub_neo4j_cdm.py

@@ -6,7 +6,7 @@ import os
 
 broker_hostname=str(os.getenv('mos_host',default="localhost"))
 port = int(os.getenv('mos_port',default=1883)) 
-client = mqtt.Client("Client2")
+client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1,"Client2")
 neo4j_uri =str(os.getenv('neo4j_host',default="bolt://localhost:7687"))
 neo4j_auth=None
 abort_time_limit = int(os.getenv('abort_time_limit', default=99999))

code/infrastructure/streaming/clients/sub/ongdb/sub_ongdb.py

@@ -6,7 +6,7 @@ import os
 
 broker_hostname=str(os.getenv('mos_host',default="localhost"))
 broker_port = int(os.getenv('mos_port',default=1883)) 
-client = mqtt.Client("Client5")
+client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1,"Client5")
 db_uri =str(os.getenv('db_host',default="bolt://localhost:7687"))
 neo4j_auth=("","")
 abort_time_limit = int(os.getenv('abort_time_limit', default=99999))

code/infrastructure/streaming/clients/sub/postgres/sub_pg_cdm.py

@@ -13,7 +13,7 @@ broker_hostname=str(os.getenv('mos_host',default="localhost"))
 broker_port = int(os.getenv('mos_port',default=1883)) 
 file_path = str(os.getenv('path_sql_script',default='C:\\Studium_MIN\\05_Masterarbeit\\thesis\\ma_code\\code\\infrastructure\\streaming\\clients\\sub\\postgres\\import_node_edge.txt'))
 
-client = mqtt.Client("Client3")
+client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1,"Client3")
 abort_time_limit = int(os.getenv('abort_time_limit', default=99999))
 
 def flatten_obj(key, val, target):