v0.1.1

-- added automativ queue reconenction (querying for queues to join from
server)

Dockerfile_build/source/modules/rabbitmq.py

@@ -16,7 +16,7 @@ from connections.quantum_backend import (
 )
 from connections.rabbitmq import rabbitmq_manager
 from fastcore.xtras import datetime
-from modules.vqe import prepare_data, run_vqe
+from modules.vqe import prepare_data, run_computation
 from pennylane.devices import Device
 from uvicorn.main import logger
 
@@ -155,7 +155,7 @@ def run_vqe_process(conn: Connection, dev: Device, data: dict):
     """Run VQE in a separate process - process exits when this function returns."""
     try:
         # Run VQE
-        run_vqe(conn=conn, dev1=dev, data=data)
+        run_computation(conn=conn, dev1=dev, data=data)
 
         # Send sentinel to indicate completion
         conn.send(None)  # Signal completion
@@ -179,6 +179,10 @@ def run_vqe_process(conn: Connection, dev: Device, data: dict):
         conn.close()
 
 
+# Global variable to track current subscriptions
+_current_subscriptions = {}  # Format: {queue_name: {"team_id": team_id, "qubits": qubits, "consumer": consumer, "queue": queue}}
+
+
 async def consume_messages_topic():
     """
     Subscribe to team-specific qubit queues.
@@ -186,6 +190,8 @@ async def consume_messages_topic():
     where N is the number of qubits this system can handle (1 to max_qubits).
 
     This enables round-robin task distribution among systems with sufficient qubits.
+
+    Dynamically manages subscriptions: adds new queues and removes obsolete ones.
     """
     channel = await rabbitmq_manager.get_consumer_channel()
     if not channel:
@@ -199,36 +205,114 @@ async def consume_messages_topic():
     access_token = get_valid_access_token()
     device_data = get_device_by_id(system_id, access_token)
     teams = device_data["teams"]
+
     if not teams:
         print(f"System {system_id} is not part of any team. No queues to subscribe.")
+        # Unsubscribe from all existing queues if no teams
+        await _unsubscribe_all()
+        return
 
-    # Subscribe to qubit-specific queues for each team
-    subscription_count = 0
+    # Build expected queues from current teams
+    expected_queues = set()
     for team in teams:
         for qubits in range(1, team["num_qubits"] + 1):
-            # Queue name format: team_{team_id}.qubits_{qubits}
             queue_name = f"team_{team['team']['team_id']}.qubits_{qubits}"
+            expected_queues.add(queue_name)
 
-            # Declare the queue (durable, shared among multiple consumers)
-            queue = await channel.declare_queue(
-                queue_name,
-                durable=True,
-                arguments={
-                    "x-max-priority": 100,  # Allow priorities 0-10
-                },
-            )
+    # Get current subscriptions
+    current_queues = set(_current_subscriptions.keys())
 
-            # Start consuming from this queue
-            await queue.consume(
-                process_message,
-                arguments={"x-priority": qubits, "x-priority-max": 10},
-            )
-            subscription_count += 1
+    # Queues to add (in expected but not current)
+    queues_to_add = expected_queues - current_queues
 
-            print(f"System {system_id} subscribed to queue: {queue_name}")
+    # Queues to remove (in current but not expected)
+    queues_to_remove = current_queues - expected_queues
 
-    print(f"System {system_id} subscribed to {subscription_count} queues")
-    await asyncio.Future()  # Keep running
+    # Unsubscribe from queues that no longer exist
+    for queue_name in queues_to_remove:
+        subscription_info = _current_subscriptions[queue_name]
+        queue = subscription_info["queue"]
+        cons = subscription_info["consumer"]
+        try:
+            # Cancel the consumer directly on the consumer object
+            await queue.cancel(consumer_tag=cons)
+            print(f"System {system_id} unsubscribed from queue: {queue_name}")
+            del _current_subscriptions[queue_name]
+        except Exception as e:
+            print(f"Error unsubscribing from {queue_name}: {e}")
+
+    # Subscribe to new queues
+    for queue_name in queues_to_add:
+        # Extract qubits from queue name
+        qubits = int(queue_name.split(".qubits_")[1])
+        team_id = queue_name.split(".qubits_")[0].replace("team_", "")
+
+        # Declare the queue (durable, shared among multiple consumers)
+        queue = await channel.declare_queue(
+            queue_name,
+            durable=True,
+            arguments={
+                "x-max-priority": 100,  # Allow priorities 0-10
+            },
+        )
+
+        # Start consuming from this queue
+        # In aio-pika, queue.consume returns a Consumer object
+        consumer = await queue.consume(
+            process_message,
+            arguments={"x-priority": qubits, "x-priority-max": 10},
+        )
+
+        # Store subscription info with consumer object
+        _current_subscriptions[queue_name] = {
+            "team_id": team_id,
+            "qubits": qubits,
+            "consumer": consumer,
+            "queue": queue,
+        }
+
+        print(f"System {system_id} subscribed to queue: {queue_name}")
+
+    # Print summary
+    if queues_to_add or queues_to_remove:
+        print(
+            f"System {system_id} subscription update - Active: {len(_current_subscriptions)} queues, "
+            f"Added: {len(queues_to_add)}, Removed: {len(queues_to_remove)}"
+        )
+    else:
+        print(
+            f"System {system_id} subscriptions unchanged - Active: {len(_current_subscriptions)} queues"
+        )
+
+
+async def _unsubscribe_all():
+    """
+    Helper function to unsubscribe from all queues.
+    """
+    for queue_name, subscription_info in list(_current_subscriptions.items()):
+        consumer = subscription_info["consumer"]
+        try:
+            # Cancel the consumer directly on the consumer object
+            await consumer.cancel()
+            print(f"Unsubscribed from queue: {queue_name}")
+            del _current_subscriptions[queue_name]
+        except Exception as e:
+            print(f"Error unsubscribing from {queue_name}: {e}")
+
+
+def get_current_subscriptions():
+    """
+    Helper function to get current subscriptions for debugging.
+    """
+    # Return a serializable version without the consumer and queue objects
+    return {
+        queue_name: {
+            "team_id": info["team_id"],
+            "qubits": info["qubits"],
+            "consumer_tag": info["consumer"].consumer_tag if info["consumer"] else None,
+        }
+        for queue_name, info in _current_subscriptions.items()
+    }
 
 
 # --- Heartbeat Publisher Logic ---