for now - this sends message to queue, but didn't actually check listeners, needs to be added

Author: tcnichol

backend/app/routers/files.py

@@ -2,7 +2,9 @@
 import time
 from datetime import datetime, timedelta
 from typing import List, Optional, Union
-
+import json
+from json import JSONEncoder
+from aio_pika import Message
 from app import dependencies
 from app.config import settings
 from app.db.file.download import _increment_file_downloads
@@ -37,6 +39,13 @@
 router = APIRouter()
 security = HTTPBearer()
 
+class CustomJSONEncoder(JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj, PydanticObjectId):
+            return str(obj)
+        # Handle other non-serializable types if needed
+        return super().default(obj)
+
 
 async def _resubmit_file_extractors(
     file: FileOut,
@@ -135,6 +144,20 @@ async def add_file_entry(
     # Add entry to the file index
     await index_file(es, FileOut(**new_file.dict()))
 
+    # Publish a message when indexing is complete
+
+    message_body = {
+        "event_type": "file_indexed",
+        "file_data": json.loads(new_file.json()),  # This handles ObjectID serialization
+        "timestamp": datetime.now().isoformat()
+    }
+
+    rabbitmq_client.basic_publish(
+        exchange='clowder',
+        routing_key='file_indexed_events',
+        body=json.dumps(message_body).encode('utf-8')
+    )
+
     # TODO - timing issue here, check_feed_listeners needs to happen asynchronously.
     time.sleep(1)
 
@@ -163,6 +186,18 @@ async def add_local_file_entry(
 
     # Add entry to the file index
     await index_file(es, FileOut(**new_file.dict()))
+    # Publish a message when indexing is complete
+    message_body = {
+        "event_type": "file_indexed",
+        "file_data": json.loads(new_file.json()),  # This handles ObjectID serialization
+        "timestamp": datetime.now().isoformat()
+    }
+
+    rabbitmq_client.basic_publish(
+        exchange='clowder',
+        routing_key='file_indexed_events',
+        body=json.dumps(message_body).encode('utf-8')
+    )
 
     # TODO - timing issue here, check_feed_listeners needs to happen asynchronously.
     time.sleep(1)

backend/message_listener.py

@@ -93,6 +93,9 @@ async def callback(message: AbstractIncomingMessage):
     async with message.process():
         msg = json.loads(message.body.decode("utf-8"))
 
+        if "event_type" in msg and msg["event_type"] == "file_indexed":
+            print(f"This is an event type file indexed!")
+
         job_id = msg["job_id"]
         message_str = msg["status"]
         timestamp = datetime.strptime(
@@ -222,6 +225,7 @@ async def listen_for_messages():
 
 
 if __name__ == "__main__":
+    logger.info(" Message listener starting...")
     start = datetime.now()
     while time_ran < timeout:
         try:

docker-compose.dev.yml

@@ -157,7 +157,7 @@ services:
       - rabbitmq
 
   extractors-messages:
-    image: "clowder/clowder2-messages:latest"
+    image: "clowder2-messages:test"
     build:
       dockerfile: backend/messages.Dockerfile
     environment: