this might work

Author: tcnichol

backend/app/dependencies.py

@@ -83,6 +83,8 @@ def get_rabbitmq() -> BlockingChannel:
     logger.debug("Connecting to rabbitmq at %s", settings.RABBITMQ_HOST)
     connection = pika.BlockingConnection(parameters)
     channel = connection.channel()
+    print(f"DEBUG: get_rabbitmq() called. Returning channel of type: {type(channel)}")
+    print(f"DEBUG: Channel object: {channel}")
     return channel
 
 

backend/app/rabbitmq/listeners.py

@@ -17,10 +17,32 @@
 from app.routers.users import get_user_job_key
 from fastapi import Depends
 from pika.adapters.blocking_connection import BlockingChannel
+import aio_pika
+from aio_pika.abc import AbstractChannel
 
 
-async def create_reply_queue():
-    channel: BlockingChannel = dependencies.get_rabbitmq()
+async def create_reply_queue(channel: AbstractChannel):
+    if (config_entry := await ConfigEntryDB.find_one({"key": "instance_id"})) is not None:
+        instance_id = config_entry.value
+    else:
+        instance_id = "".join(
+            random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits)
+            for _ in range(10)
+        )
+        config_entry = ConfigEntryDB(key="instance_id", value=instance_id)
+        await config_entry.insert()
+
+    queue_name = f"clowder.{instance_id}"
+
+    # Use aio_pika methods instead of pika methods
+    exchange = await channel.declare_exchange("clowder", durable=True)
+    queue = await channel.declare_queue(queue_name, durable=True, exclusive=False, auto_delete=False)
+    await queue.bind(exchange)
+
+    return queue.name
+
+async def create_reply_queue(channel: BlockingChannel):
+    # channel: BlockingChannel = dependencies.get_rabbitmq()
 
     if (
         config_entry := await ConfigEntryDB.find_one({"key": "instance_id"})
@@ -52,8 +74,12 @@ async def submit_file_job(
     routing_key: str,
     parameters: dict,
     user: UserOut,
-    rabbitmq_client: BlockingChannel,
+    rabbitmq_client: AbstractChannel,
 ):
+    # print(f"DEBUG submit_file_job: Got client of type: {type(rabbitmq_client)}")
+    # if not isinstance(rabbitmq_client, BlockingChannel):
+    #     raise TypeError(f"Expected BlockingChannel, got {type(rabbitmq_client)}. This confirms a mixing issue.")
+
     # Create an entry in job history with unique ID
     job = EventListenerJobDB(
         listener_id=routing_key,
@@ -65,6 +91,7 @@ async def submit_file_job(
     )
     await job.insert()
 
+
     current_secretKey = await get_user_job_key(user.email)
     msg_body = EventListenerJobMessage(
         filename=file_out.name,
@@ -75,15 +102,20 @@ async def submit_file_job(
         job_id=str(job.id),
         parameters=parameters,
     )
-    reply_to = await create_reply_queue()
+
+    # Use aio_pika publishing
+    # Get the existing clowder exchange
+    exchange = await rabbitmq_client.get_exchange("clowder")
+    reply_to = await create_reply_queue(rabbitmq_client)
     print("RABBITMQ_CLIENT: " + str(rabbitmq_client))
-    rabbitmq_client.basic_publish(
-        exchange="",
-        routing_key=routing_key,
-        body=json.dumps(msg_body.dict(), ensure_ascii=False),
-        properties=pika.BasicProperties(
-            content_type="application/json", delivery_mode=1, reply_to=reply_to
+    await exchange.publish(
+        aio_pika.Message(
+            body=json.dumps(msg_body.dict(), ensure_ascii=False).encode('utf-8'),
+            content_type="application/json",
+            delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+            reply_to=reply_to,
         ),
+        routing_key=routing_key,
     )
     return str(job.id)